Repository: cloudspannerecosystem/harbourbridge
Pull request #225 (2021-11-18 07:17:56)

Security patch for dependabot alerts
There are 2 dependabot alerts for the backend libs:

- **github.com/gorilla/websocket :**  a version bump should fix.
- **github.com/dgrijalva/jwt-go :** This library is unmaintained hence no version bump available. This is imported through the **pingcap/tidb** library. There are 2 ways to go about this:
   1. **(Did not work)** Bump TiDB to latest version (that has the fix for this issue): This cannot be done because pingcap/tidb does not version as per golang recommendations i.e. they do not append /v2, /v3 , /v4 ... at the end of their module paths. Hence we cannot bump the version up.
   2. **(Approach taken)** TiDB replaced this lib with a fork named form3tech-oss that has the fix in their latest version. We put the same replace directive in our go.mod without bumping the TiDB version. 
  
**Caveat to the approach taken:**
The form3tech-oss fork has the fix but is unmaintained now. The ideal solution would be to switch to a community maintained fork named golang-jwt but TiDB still uses the former fork hence keeping it same as that.


Patch:
diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml
index 7104b4b..16d5aa6 100644
--- a/.github/workflows/lint.yaml
+++ b/.github/workflows/lint.yaml
@@ -16,3 +16,4 @@ jobs:
       - run: go vet .
       - run: go get golang.org/x/tools/cmd/goimports && test -z $(goimports -l .)
       - run: go get golang.org/x/lint/golint && golint .
+      - run: go get honnef.co/go/tools/cmd/staticcheck && staticcheck .
diff --git a/README.md b/README.md
index ecbd275..afd860f 100644
--- a/README.md
+++ b/README.md
@@ -3,8 +3,8 @@
 [![cloudspannerecosystem](https://circleci.com/gh/cloudspannerecosystem/harbourbridge.svg?style=svg)](https://circleci.com/gh/cloudspannerecosystem/harbourbridge)
 
 HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation and
-migration, using data from an existing PostgreSQL, MySQL or DynamoDB database. 
-The tool ingests schema and data from either a pg_dump/mysqldump file or directly 
+migration, using data from an existing PostgreSQL, MySQL or DynamoDB database.
+The tool ingests schema and data from either a pg_dump/mysqldump file or directly
 from the source database, and supports both schema and data migration. For schema
 migration, HarbourBridge automatically builds a Spanner schema from the schema
 of the source database. This schema can be customized using the HarbourBridge
@@ -17,13 +17,13 @@ For more details on schema customization and use of the schema assistant, see
 capabilities of HarbourBridge.
 
 HarbourBridge is designed to simplify Spanner evaluation and migration, and in
-particular for migrating moderate-size PostgreSQL/MySQL/DynamoDB datasets to Spanner 
+particular for migrating moderate-size PostgreSQL/MySQL/DynamoDB datasets to Spanner
 (up to about 100GB). Many features of PostgreSQL/MySQL, especially those that don't
 map directly to Spanner features, are ignored, e.g. stored functions and
 procedures, and sequences. Types such as integers, floats, char/text, bools,
 timestamps, and (some) array types, map fairly directly to Spanner, but many
 other types do not and instead are mapped to Spanner's `STRING(MAX)`.
-In the case of DynamoDB, the schema is inferred based on a certain 
+In the case of DynamoDB, the schema is inferred based on a certain
 amount of sampled data.
 
 View HarbourBridge as a way to get up and running fast, so you can focus on
@@ -31,34 +31,13 @@ critical things like tuning performance and getting the most out of
 Spanner. Expect that you'll need to tweak and enhance what HarbourBridge
 produces.
 
-To use the tool on a PostgreSQL database called mydb, run
-
-```sh
-# By default, the driver is "pg_dump".
-pg_dump mydb | harbourbridge
-# Or,
-pg_dump mydb | harbourbridge -driver=pg_dump
-```
-
-To use the tool on a MySQL database called mydb, run
-
-```sh
-mysqldump mydb | harbourbridge -driver=mysqldump
-```
-
-To use the tool on a DynamoDB database, run
-```sh
-harbourbridge -driver=dynamodb
-```
+For some quick starter examples on how to run HarbourBridge, take a look at
+[Quickstart Guide](#quickstart-guide).
 
-HarbourBridge accepts pg_dump/mysqldump's standard plain-text format, but 
-not archive or custom formats. More details on usage can be found in 
-[Example usage](#example-usage) section.
-
-HarbourBridge automatically determines the cloud project and Spanner instance to
-use, and generates a new Spanner database name (prefixed with `{driver}_` and
-today's date). Command-line flags can be used to explicitly set the Spanner
-instance or database name. See [Options](#options).
+HarbourBridge automatically determines the cloud project to use, and generates
+a new Spanner database name.
+Command-line flags can be used to explicitly set the Spanner instance or
+database name. See [Command line flags](#command-line-flags).
 
 **WARNING: Please check that permissions for the Spanner instance used by
 HarbourBridge are appropriate. Spanner manages access control at the database
@@ -117,21 +96,22 @@ Spanner instance.
 
 Install Go ([download](https://golang.org/doc/install)) on your development
 machine if it is not already installed, configure the GOPATH environment
-variable if it is not already configured using 
-`export GOPATH=$(go env GOPATH)`, and 
+variable if it is not already configured, and
 [test your installation](https://golang.org/doc/install#testing).
 
 ### Installing HarbourBridge
 
 You can make a copy of the HarbourBridge codebase from the github repository
-and use "go run":
+and use "go run".
 
 ```sh
 git clone https://github.com/cloudspannerecosystem/harbourbridge
 cd harbourbridge
-go run github.com/cloudspannerecosystem/harbourbridge -driver=pg_dump < my_pg_dump_file
+go run github.com/cloudspannerecosystem/harbourbridge help
 ```
 
+Examples below assume that `harbourbridge` alias is set as following
+
 ```sh
 alias harbourbridge="go run github.com/cloudspannerecosystem/harbourbridge"
 ```
@@ -143,34 +123,43 @@ This workflow also allows you to modify or customize the HarbourBridge codebase.
 To use the tool on a PostgreSQL database called mydb, run
 
 ```sh
-pg_dump mydb | harbourbridge -driver=pg_dump
+pg_dump mydb > mydb.pg_dump
+harbourbridge schema-and-data -source=postgresql < mydb.pg_dump
 ```
 
 To use the tool on a MySQL database called mydb, run
 
 ```sh
-mysqldump mydb | harbourbridge -driver=mysqldump
+mysqldump mydb > mydb.mysqldump
+harbourbridge schema-and-data -source=mysql < mydb.mysqldump
 ```
 
 To use the tool on a DynamoDB database, run
 
 ```sh
-harbourbridge -driver=dynamodb
+harbourbridge schema-and-data -source=dynamodb
 ```
-More details on running harbourbridge can be found in [Example usage](#example-usage) section.
+
+Note: HarbourBridge accepts pg_dump/mysqldump's standard plain-text format,
+but not archive or custom formats. More details on HarbourBridge example usage
+can be found here:
+
+- [PostgreSQL example usage](sources/postgres/README.md#example-postgresql-usage)
+- [MySQL example usage](sources/mysql/README.md#example-mysql-usage)
+- [DynamoDB example usage](sources/dynamodb/README.md#example-dynamodb-usage)
 
 This command will use the cloud project specified by the `GCLOUD_PROJECT`
 environment variable, automatically determine the Cloud Spanner instance
-associated with this project, convert the source schema to a Spanner schema 
+associated with this project, convert the source schema to a Spanner schema
 (For MySQL/Postgres) or infer a schema from the DynamoDB instance, create a
 new Cloud Spanner database with this schema, and finally, populate this new
-database with the data from the source database. 
-If the project has multiple instances, then list of available instances 
-will be shown and you will have to pick one of the available instances and 
-set the `--instance` flag. The new Cloud Spanner database will have a name of 
-the form `{DRIVER}_{DATE}_{RANDOM}`, where`{DRIVER}` is the drivername 
-specified,`{DATE}`is today's date, and`{RANDOM}` is a random suffix for 
-uniqueness.
+database with the data from the source database.
+If the project has multiple instances, then list of available instances
+will be shown and you will have to pick one of the available instances and
+set the `--instance` flag in target-profile. The new Cloud Spanner database
+will have a name of the form `{SOURCE}_{DATE}_{RANDOM}`, where`{SOURCE}` is the
+value of the source flag,`{DATE}`is today's date, and`{RANDOM}` is a random
+suffix for uniqueness.
 
 See the [Troubleshooting Guide](#troubleshooting-guide) for help on debugging
 issues.
@@ -190,11 +179,10 @@ and one for user carts). The files [singers.pg_dump](examples/singers.pg_dump)
 and [singers.mysqldump](examples/singers.mysqldump) contain pg_dump and
 mysqldump output for a version of the [Cloud Spanner
 singers](https://cloud.google.com/spanner/docs/schema-and-data-model#creating_a_table)
-example. To use HarbourBridge on cart.pg_dump, download the file locally and
-run
+example. To use HarbourBridge on cart.pg_dump, download the file locally and run
 
-```
-harbourbridge -driver=pg_dump < cart.pg_dump
+```sh
+harbourbridge schema -source=postgresql < cart.pg_dump
 ```
 
 ### Verifying Results
@@ -207,7 +195,7 @@ it. This will list the tables created by HarbourBridge. Select a table, and take
 a look at its schema and data. Next, go to the query page, and try
 some SQL statements. For example
 
-```
+```sql
 SELECT COUNT(*) from mytable
 ```
 
@@ -217,13 +205,13 @@ to check the number of rows in table `mytable`.
 
 The tables created by HarbourBridge provide a starting point for evaluation of
 Spanner. While they preserve much of the core structure of your PostgreSQL/MySQL
-schema and data, many key features have been dropped, including functions, 
-sequences, procedures,triggers, and views. For DynamoDB, the conversion from 
-schemaless to schema is focused on the use-case where customers use DynamoDB in 
+schema and data, many key features have been dropped, including functions,
+sequences, procedures,triggers, and views. For DynamoDB, the conversion from
+schemaless to schema is focused on the use-case where customers use DynamoDB in
 a consistent, structured way with a fairly well defined set of columns and types.
 
 As a result, the out-of-the-box performance you get from these tables could be
-slower than what you get from PostgreSQL/MySQL/DynamoDB. 
+slower than what you get from PostgreSQL/MySQL/DynamoDB.
 
 To improve performance, also consider using [Interleaved
 Tables](https://cloud.google.com/spanner/docs/schema-and-data-model#creating-interleaved-tables)
@@ -240,11 +228,11 @@ plans and aspirations for developing HarbourBridge further are outlined in the
 Whitepaper](https://github.com/cloudspannerecosystem/harbourbridge/blob/master/whitepaper.md).
 
 You can also change the way HarbourBridge behaves by directly editing the
-pg_dump/mysqldump output. For example, suppose you want to try out different 
-primary keys for a table. First run pg_dump/mysqldump and save the output to 
-a file. Then modify (or add) the relevant 
-`ALTER TABLE ... ADD CONSTRAINT ... PRIMARY KEY ...` statement in the 
-pg_dump/mysqldump output file so that the primary keys match what you need. 
+pg_dump/mysqldump output. For example, suppose you want to try out different
+primary keys for a table. First run pg_dump/mysqldump and save the output to
+a file. Then modify (or add) the relevant
+`ALTER TABLE ... ADD CONSTRAINT ... PRIMARY KEY ...` statement in the
+pg_dump/mysqldump output file so that the primary keys match what you need.
 Then run HarbourBridge on the modified pg_dump/mysqldump output.
 
 ## Files Generated by HarbourBridge
@@ -260,8 +248,8 @@ HarbourBridge generates several files as it runs:
 
 - Report file (ending in `report.txt`): contains a detailed analysis of the
   PostgreSQL/MySQL to Spanner migration, including table-by-table stats and an
-  analysis of PostgreSQL/MySQL types that don't cleanly map onto Spanner types. 
-  Note that PostgreSQL/MySQL types that don't have a corresponding Spanner type 
+  analysis of PostgreSQL/MySQL types that don't cleanly map onto Spanner types.
+  Note that PostgreSQL/MySQL types that don't have a corresponding Spanner type
   are mapped to STRING(MAX).
 
 - Bad data file (ending in `dropped.txt`): contains details of data
@@ -273,75 +261,132 @@ By default, these files are prefixed by the name of the Spanner database (with a
 dot separator). The file prefix can be overridden using the `-prefix`
 [option](#options).
 
-## Options
+## HarbourBridge CLI (command line interface)
 
-HarbourBridge accepts the following options:
+HarbourBridge CLI follows [subcommands](https://github.com/google/subcommands)
+structure with the the following general syntax:
 
-`-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.
+```sh
+harbourbridge <subcommand> flags
+```
 
-`-instance` Specifies the Spanner instance to use. The new database will be
-created in this instance. If not specified, the tool automatically determines an
-appropriate instance using gcloud.
+### Getting Help
+
+The command `harbourbridge help` displays the available subcommands and the important global flags.
+
+```text
+    commands   list all subcommand names
+    help   describe subcommands and their syntax
+```
+
+To get help on individual subcommands, use
+
+```sh
+    harbourbridge help <subcommand>
+```
+
+This will print the usage pattern, a few examples, and a list of all available subcommand flags.
+
+### Subcommands
+
+#### harbourbridge `schema`
+
+This subcommand can be used to perform schema migration and report on the quality of the migration. Generated schema mapping file (session.json) can be then further edited using the HarbourBridge web UI to make custom edits to the destination schema. This session file
+is then passed to the data subcommand to perform data migration while honoring the defined
+schema mapping. HarbourBridge also generates Spanner schema which users can modify manually and use directly as well.
 
-`-driver` Specifies the driver to use for schema and data conversion. Supported 
-drivers are _'postgres'_, _'pg_dump'_, _'mysql'_ and _'mysqldump'_. By default, 
-the driver is _'pg_dump'_.
+#### harbourbridge `data`
 
-`-schema-sample-size` Specifies the number of rows to use for inferring schema 
-(only for DynamoDB). By default, the schema sample size is 100,000.
+This subcommand will perform data migration and report on the quality of the same. Rows which could not be migrated are reported in
+dropped.txt file. This subcommand requires users to pass the session file (which contains schema mapping) generated by either the `schema` subcommand or web UI.
+
+#### harbourbridge `schema-and-data`
+
+This subcommand will generate a schema as well as perform data migration and report on the quality of both schema migration and data migration. This subcommand can be used to do a quick evaluation for the migration and get started quickly on Spanner.
+
+### Command line flags
+
+This section describes the flags common across all the subcommands. For flags
+specific to a give subcommand run `harbourbridge help <subcommand>`.
+
+`-source` Required flag. Specifies the source database. Supported source
+databases are _'postgres'_, _'mysql'_ and _'dynamodb'_.
+
+`-target` Optional flag. Specifies the target database. Defaults to _'spanner'_
+, which is the only supported target database today.
 
 `-prefix` Specifies a file prefix for the report, schema, and bad-data files
 written by the tool. If no file prefix is specified, the name of the Spanner
 database (plus a '.') is used.
 
-`-v` or `-verbose` Specifies verbose mode. This will cause HarbourBridge to output detailed
-messages about the conversion.
-
-`-schema-only` Specifies that only schema processing will be performed.
-Any data in the source database will be ignored.
-
-`-data-only` Specifies that only data migration will be performed.
-A spanner database will be created based on the schema state provided
-by a session file (`-session`) and data will be migrated.
+`-v` or `-verbose` Specifies verbose mode. This will cause HarbourBridge to
+output detailed messages about the conversion.
 
 `-skip-foreign-keys` Controls whether we add foreign key constraints after
 data migration is complete. This flag cannot be used with schema-only mode,
 and does not affect the generation of foreign key statements during schema
-processing i.e. foreign key constraints will still appear in the generated 
+processing i.e. foreign key constraints will still appear in the generated
 Spanner DDL files.
 
-`-session` Specifies a session file that contains all schema and data 
+`-session` Specifies a session file that contains all schema and data
 conversion state endcoded as JSON.
 
-`-dump-file` Specifies the path where the dump file is stored. 
-If the location is Google Cloud Storage (GCS), you can use the following format:
-`-dump-file=gs://{bucket_name}/{path/to/dump_file}`. Please ensure you have
-read pemissions to the GCS bucket you would like to use. 
+`-source-profile` Specifies detailed parameters for the source database such as connection parameters. See [Source Profile](#source-profile) for details.
 
-## Example Usage
+`-target-profile` Specifies detailed parameters for the target database. See [Target Profile](#target-profile) for details.
 
-Details on HarbourBridge example usage can be found here: 
-- [PostgreSQL example usage](sources/postgres/README.md#example-postgresql-usage)
-- [MySQL example usage](sources/mysql/README.md#example-mysql-usage)
-- [DynamoDB example usage](sources/dynamodb/README.md#example-dynamodb-usage)
+### Source Profile
+
+HarbourBridge accepts the following params for --source-profile,
+specified as "key1=value1,key2=value,..." pairs:
+
+`file` Specifies the full path of the file to use for reading source database
+schema and/or data. This param is optional, and the file can also be piped to
+stdin, if available locally.
+
+If the file is located in Google Cloud Storage (GCS), you can use the
+following format: `file=gs://{bucket_name}/{path/to/file}`. Please ensure you
+have read pemissions to the GCS bucket you would like to use.
+
+`format` Specifies the format of the file. This param is also optional, and
+defaults to `dump`. This may be extended in future to support other formats
+such as `csv`, `avro` etc.
+
+### Target Profile
+
+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.
 
+`instance` Specifies the Spanner instance to use. The new database will be
+created in this instance. If not specified, the tool automatically determines an
+appropriate instance using gcloud.
+
+`dialect` Specifies the dialect of Spanner database. By default, Spanner
+databases are created with GoogleSQL dialect. You can override the same by
+setting `dialect=PostgreSQL` in the `-target-profile`. Learn more about support
+for PostgreSQL dialect in Cloud Spanner at
+<https://cloud.google.com/spanner/docs/postgresql-interface>.
 
 ## Schema Conversion
 
 Details on HarbourBridge schema conversion can be found here:
+
 - [PostgreSQL schema conversion](sources/postgres/README.md#schema-conversion)
 - [MySQL schema conversion](sources/mysql/README.md#schema-conversion)
 - [DynamoDB schema conversion](sources/dynamodb/README.md#schema-conversion)
 
 ## Data Conversion
 
-HarbourBridge converts PostgreSQL/MySQL/DynamoDB data to Spanner data based on 
-the Spanner schema it constructs. Conversion for most data types is fairly 
-straightforward, but several types deserve discussion. Note that HarbourBridge 
-is not intended for databases larger than about 100GB. Details on HarbourBridge 
+HarbourBridge converts PostgreSQL/MySQL/DynamoDB data to Spanner data based on
+the Spanner schema it constructs. Conversion for most data types is fairly
+straightforward, but several types deserve discussion. Note that HarbourBridge
+is not intended for databases larger than about 100GB. Details on HarbourBridge
 data conversion can be found here:
+
 - [PostgreSQL data conversion](sources/postgres/README.md#data-conversion)
 - [MySQL data conversion](sources/mysql/README.md#data-conversion)
 - [DynamoDB data conversion](sources/dynamodb/README.md#data-conversion)
@@ -351,14 +396,29 @@ data conversion can be found here:
 The following steps can help diagnose common issues encountered while running
 HarbourBridge.
 
-### 1. Verify driver configuration
+### 1. Verify source profile configuration
+
+First, check that the source profile is correctly configured to connect to your
+database. Source profile configuration varies depending on the database.
+
+#### 1.1 Direct access to PostgreSQL
+
+See [Directly connecting to a PostgreSQL database](sources/postgres/README.md#directly-connecting-to-a-postgresql-database) for troubleshooting direct access to PostgreSQL.
+
+#### 1.2 Direct access to MySQL
+
+See [Directly connecting to a MySQL database](sources/mysql/README.md#directly-connecting-to-a-mysql-database) for troubleshooting direct access to MySQL.
 
-First, check that the driver is correctly configured to connect to your
-database. Driver configuration varies depending on the driver used.
+#### 1.3 Direct access to DynamoDB
 
-#### 1.1 pg_dump
+See [DynamoDB example usage](sources/dynamodb/README.md#example-dynamodb-usage)
+for troubleshooting direct access to DynamoDB.
 
-If you are using pg_dump (-driver=pg_dump), check that pg_dump is
+### 2. Generating dump files
+
+#### 2.1 Generating pg_dump file
+
+If you are using pg_dump , check that pg_dump is
 correctly configured to connect to your PostgreSQL
 database. Note that pg_dump uses the same options as psql to connect to your
 database. See the [psql](https://www.postgresql.org/docs/9.3/app-psql.html) and
@@ -371,45 +431,17 @@ which are standard across PostgreSQL utilities.
 It is also possible to configure access via pg_dump's command-line options
 `--host`, `--port`, and `--username`.
 
-#### 1.2 Direct access to PostgreSQL
-
-In this case, HarbourBridge connects directly to the PostgreSQL
-database to retrieve table schema and data. Set the `-driver=postgres`
-and provide the environment variables `PGHOST`, `PGPORT`, `PGUSER`, 
-`PGDATABASE`. Password can be specified either in the `PGPASSWORD`
-environment variable or provided at the password prompt.
-
-#### 1.3 mysqldump
-
-If you are using mysqldump (-driver=mysqldump), check that mysqldump is
-correctly configured to connect to your MySQL
-database via the command-line options `--host`, `--port`, and `--user`.
-Note that mysqldump uses the same options as mysql to connect to your
-database. See the [mysql](https://dev.mysql.com/doc/refman/8.0/en/mysql-commands.html)
-and [mysqldump](https://dev.mysql.com/doc/refman/8.0/en/mysqldump.html) documentation.
-
-#### 1.4 Direct access to MySQL
-
-In this case, HarbourBridge connects directly to the MySQL
-database to retrieve table schema and data. Set the `-driver=mysql`
-and provide the environment variables `MYSQLHOST`, `MYSQLPORT`, 
-`MYSQLUSER`, `MYSQLDATABASE`. Password can be specified either 
-in the `MYSQLPWD` environment variable or provided at the password 
-prompt.
-
-
-#### 1.5 Direct access to DynamoDB
-
-In this case, HarbourBridge connects directly to the DynamoDB
-database to retrieve table schema and data. Set the `-driver=dynamodb`
-and provide the environment variables `AWS_ACCESS_KEY_ID`, 
-`AWS_SECRET_ACCESS_KEY`, `AWS_REGION`. If you use a custom endpoint for 
-dynamodb, you can specify that using the environment variable 
-`DYNAMODB_ENDPOINT_OVERRIDE`.
+#### 2.2 mysqldump
 
-### 2. Verify dump output
+If you are using mysqldump, check that mysqldump is correctly configured to
+connect to your MySQL database via the command-line options `--host`, `--port`,
+and `--user`. Note that mysqldump uses the same options as mysql to connect to
+your database. See the
+[mysql](https://dev.mysql.com/doc/refman/8.0/en/mysql-commands.html) and
+[mysqldump](https://dev.mysql.com/doc/refman/8.0/en/mysqldump.html)
+documentation.
 
-Next, verify that pg_dump/mysqldump is generating plain-text output. If your 
+Next, verify that pg_dump/mysqldump is generating plain-text output. If your
 database is small, try running
 
 ```sh
@@ -420,10 +452,10 @@ and look at the output file. It should be a plain-text file containing SQL
 commands. If your database is large, consider just dumping the schema via the
 `--schema-only` for pg_dump and `--no-data` for mysqldump command-line option.
 
-pg_dump/mysqldump can export data in a variety of formats, but HarbourBridge 
+pg_dump/mysqldump can export data in a variety of formats, but HarbourBridge
 only accepts `plain` format (aka plain-text). See the
 [pg_dump documentation](https://www.postgresql.org/docs/9.3/app-pgdump.html) and
-[mysqldump documentation](https://dev.mysql.com/doc/refman/8.0/en/mysqldump.html) 
+[mysqldump documentation](https://dev.mysql.com/doc/refman/8.0/en/mysqldump.html)
 for details about formats.
 
 ### 3. Debugging HarbourBridge
@@ -432,12 +464,12 @@ The HarbourBridge tool can fail for a number of reasons.
 
 #### 3.1 No space left on device
 
-HarbourBridge needs to read the pg_dump/mysqldump output twice, once to build 
+HarbourBridge needs to read the pg_dump/mysqldump output twice, once to build
 a schema and once for data ingestion. When pg_dump/mysqldump output is directly
-piped to HarbourBridge, `stdin` is not seekable, and so we write the output to 
-a temporary file. That temporary file is created via Go's ioutil.TempFile. 
-On many systems, this creates a file in `/tmp`, which is sometimes configured 
-with minimal space. A simple workaround is to separately run pg_dump/mysqldump 
+piped to HarbourBridge, `stdin` is not seekable, and so we write the output to
+a temporary file. That temporary file is created via Go's ioutil.TempFile.
+On many systems, this creates a file in `/tmp`, which is sometimes configured
+with minimal space. A simple workaround is to separately run pg_dump/mysqldump
 and write its output to a file in a directory with sufficient space. For example,
 if the current working directory has space, then:
 
@@ -453,9 +485,9 @@ environment variable.
 #### 3.2 Unparsable dump output
 
 HarbourBridge uses the [pg_query_go](https://github.com/pganalyze/pg_query_go)
-library for parsing pg_dump and [pingcap parser](https://github.com/pingcap/parser) 
-for parsing mysqldump. It is possible that the pg_dump/mysqldump output is 
-corrupted or uses features that aren't parseable. Parsing errors should 
+library for parsing pg_dump and [pingcap parser](https://github.com/pingcap/parser)
+for parsing mysqldump. It is possible that the pg_dump/mysqldump output is
+corrupted or uses features that aren't parseable. Parsing errors should
 generate an error message of the form `Error parsing last 54321 line(s) of input`.
 
 #### 3.2 Credentials problems
@@ -463,7 +495,7 @@ generate an error message of the form `Error parsing last 54321 line(s) of input
 HarbourBridge uses standard Google Cloud credential mechanisms for accessing
 Cloud Spanner. If this is mis-configured, you may see errors containing
 "unauthenticated", or "cannot fetch token", or "could not find default
-credentials". You might need to run `gcloud auth application-default login`. 
+credentials". You might need to run `gcloud auth application-default login`.
 See the [Before you begin](#before-you-begin) section for details.
 
 #### 3.4 Can't create database
diff --git a/cmd/README.dynamodb.md b/cmd/README.dynamodb.md
deleted file mode 100644
index 881b95b..0000000
--- a/cmd/README.dynamodb.md
+++ /dev/null
@@ -1,147 +0,0 @@
-# HarbourBridge: Turnkey DynamoDB-to-Spanner Evaluation
-
-HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
-using data from an existing PostgreSQL or MySQL or DynamoDB database. This
-README provides details of the tool's DynamoDB capabilities. For general
-HarbourBridge information see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-turnkey-spanner-evaluation).
-
-## Example DynamoDB Usage
-
-The following examples assume `harbourbridge` has been added to your PATH
-environment variable.
-
-Before running HarbourBridge, make sure that you have
-[set up your AWS credentials/region](https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html)
-correctly (set the environment variables `AWS_ACCESS_KEY_ID`, 
-`AWS_SECRET_ACCESS_KEY`, `AWS_REGION`). HarbourBridge accesses your 
-DynamoDB database via the aws go sdk package. If you use a custom endpoint
-for dynamodb, you can specify that using the environment variable 
-`DYNAMODB_ENDPOINT_OVERRIDE`.
-
-```sh
-export DYNAMODB_ENDPOINT_OVERRIDE=http://dynamodb.us-west-2.amazonaws.com`
-```
-
-To use the tool on a DynamoDB database (it will migrate all tables),
-run
-
-```sh
-harbourbridge schema -source=dynamodb
-```
-
-### Source profile (`-source-profile`)
-
-HarbourBridge accepts the following options for --source-profile for DynamoDB,
-specified as "key1=value1,key2=value,..." pairs:
-
-`-schema-sample-size` Due to the schemaless nature of DynamoDB, the tool infers
-the schema based on a certain amount of sampled data, by default, 100000 rows.
-If a table has more rows than the default value, we only use 100000 rows for 
-estimating the schema. This flag lets you specify the number of rows to use for
-inferring schema. The default value is 100,000.
-
-```sh
-harbourbridge schema -source=dynamodb -source-profile="schema-sample-size=500000"
-```
-
-## Schema Conversion
-
-The HarbourBridge tool maps DynamoDB types to Spanner types as follows:
-
-| PostgreSQL Type    | Spanner Type               | Notes                                     |
-| ------------------ | -------------------------- | ----------------------------------------- |
-| `Number`           | `NUMERIC` or `STRING`      | defaults to NUMERIC, otherwise, STRING    |
-| `String`           | `STRING`                   |                                           |
-| `Boolean`          | `BOOL`                     |                                           |
-| `Binary`           | `BYTES`                    |                                           |
-| `Null`             | A nullable column type     |                                           |
-| `List`             | `STRING`                   | json encoding                             |
-| `Map`              | `STRING`                   | json encoding                             |
-| `StringSet`        | `ARRAY<STRING>`            |                                           |
-| `NumberSet`        | `ARRAY<NUMERIC or STRING>` |                                           |
-| `BinarySet`        | `ARRAY<BYTES>`             |                                           |
-
-We discuss these, as well as other limits and notes on schema conversion, in the
-following sections.
-
-### Schema Inference
-
-DynamoDB is a schemaless database: other than a primary index and optional
-secondary index, column names and types are essentially unconstrained
-and can vary from one row to the next.
-
-However, many customers use DynamoDB in a consistent, structured way
-with a fairly well defined set of columns and types. Our HarbourBridge support
-for DynamoDB focuses on this use-case, and we construct a Spanner schema
-by inspecting table data.
-
-For small tables, we inspect all rows of the table. For large tables, scanning
-the entire table would be extremely expensive and slow, and so we only inspect
-the first N rows (defined by the flag `schema-sample-size`) from the table scan.
-While DynamoDB doesn't return scan results in order, they might not be a truly
-random sample of rows. However, the alternative of randomly sampling rows
-would be much more expensive.
-
-Columns with consistent types are assigned Spanner types as detailed below.
-Columns without a consistent type are mapped to STRING.
-
-#### `Number`
-
-In most cases, we map the Number type in DynamoDB to Spanner's Numeric type.
-However, since the [range of Numeric](https://cloud.google.com/spanner/docs/storing-numeric-data)
-in Cloud Spanner is smaller than the [range of Number](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html)
-in DynamoDB, this conversion could result in out of range with potential
-precision loss. To address this possibility, we try to convert the sample data,
-and if it consistently fails, we choose STRING type for the column.
-
-#### `Null` Data Type
-
-In DynamoDB, a column can have a Null data type that represents an unknown or
-undefined state. Also, each row defines its own schema for columns (not for
-primary keys). So columns can be absent in rows. 
-
-We treat the above two cases the same as a Null value in Cloud Spanner. The
-cases that a column contains a Null value or a column is not present is an
-indication that this column should be nullable.
-
-#### `List` and `Map`
-
-In Cloud Spanner, the most similar type to List and Map is
-[STRUCT](https://cloud.google.com/spanner/docs/data-types#struct_type), but it
-is not a valid column type (available for query but not for storage).
-Therefore, we encode them into a json string. 
-
-#### Occasional Errors
-
-To prevent a few spurious rows from impacting schema construction, we define an
-error threshold: when building a type for a column, if the percentage of rows
-with a specific data type is lower than or equal to an extremely low
-value (0.1%), then we treat those rows as likely errors. Such rows are ignored
-for schema construction: their type is not considered a candidate type for the
-column.
-
-#### Multi-type Columns
-
-For a special scenario, we may get a column that has equal distribution of two
-data types. E.g., a column has 40% rows in String and 60% rows in Number. If we
-choose Number as its type, it may lead to 40% data loss in the data conversion.
-In the migration, we define a conflicting threshold on rows (after removing Null
-data types and rows that the column is not present). By default, the conflicting
-threshold is 5% and if the percentages of two or more data types are greater
-than it, we would consider that the column has conflicting data types. As a safe
-choice, we define this column as a STRING type in Cloud Spanner. 
-
-## Data Conversion
-
-### A Scan for Entire Table
-
-Data conversion proceeds table by table. For each table, we use the Scan API to
-read data. Each read has a size limit up to 1MB. By using the returned token, we
-make a subsequent call to continue retrieving data from the table.
-
-The row result contains the data type and data itself. According to our
-[inferred schema](#schema-inference), we will parse the row to a format that
-Cloud Spanner can support. If the value parsing fails, we would drop the entire
-row and record it as bad data in the report. If a column does not appear or 
-column has a NULL data type, we would process this as a NULL value in 
-Cloud Spanner. 
diff --git a/cmd/README.md b/cmd/README.md
deleted file mode 100644
index 3c9cc05..0000000
--- a/cmd/README.md
+++ /dev/null
@@ -1,527 +0,0 @@
-# HarbourBridge: Spanner Evaluation and Migration
-
-[![cloudspannerecosystem](https://circleci.com/gh/cloudspannerecosystem/harbourbridge.svg?style=svg)](https://circleci.com/gh/cloudspannerecosystem/harbourbridge)
-
-HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation and
-migration, using data from an existing PostgreSQL, MySQL or DynamoDB database. 
-The tool ingests schema and data from either a pg_dump/mysqldump file or directly 
-from the source database, and supports both schema and data migration. For schema
-migration, HarbourBridge automatically builds a Spanner schema from the schema
-of the source database. This schema can be customized using the HarbourBridge
-schema assistant. For data migration, HarbourBridge creates a new Spanner
-database using the Spanner schema built during schema migration, and populates
-it with data from the source database.
-
-For more details on schema customization and use of the schema assistant, see
-[web/README](web/README.md). The rest of this README describes the command-line
-capabilities of HarbourBridge.
-
-HarbourBridge is designed to simplify Spanner evaluation and migration, and in
-particular for migrating moderate-size PostgreSQL/MySQL/DynamoDB datasets to Spanner 
-(up to about 100GB). Many features of PostgreSQL/MySQL, especially those that don't
-map directly to Spanner features, are ignored, e.g. stored functions and
-procedures, and sequences. Types such as integers, floats, char/text, bools,
-timestamps, and (some) array types, map fairly directly to Spanner, but many
-other types do not and instead are mapped to Spanner's `STRING(MAX)`.
-In the case of DynamoDB, the schema is inferred based on a certain 
-amount of sampled data.
-
-View HarbourBridge as a way to get up and running fast, so you can focus on
-critical things like tuning performance and getting the most out of
-Spanner. Expect that you'll need to tweak and enhance what HarbourBridge
-produces.
-
-Below are some quick starter examples on how to run HarbourBridge. You can look
-at specific sections for PostgreSQL, MySQL etc for detailed examples.
-
-For example, to generate Spanner schema from a PostgreSQL dump file, run
-
-```sh
-# Generate pg_dump file.
-pg_dump mydb > mydb.pg_dump
-
-harbourbridge schema -source=postgresql -source-profile="format=dump" < mydb.pg_dump
-# Or, (By default, format is "dump" when specifying input files)
-harbourbridge schema -source=postgresql < mydb.pg_dump
-# Or, you can specify full path to the dump file
-harbourbridge schema -source=postgresql -source-profile="format=dump,file=<full path to dump file>"
-# You can also specify path to a file stored in GCS (Google Cloud Storage)
-harbourbridge schema -source=postgresql -source-profile="format=dump,file=gs://<bucket_name>/<file_path>"
-
-# Target is Spanner by default, but you can also specify it explicitly.
-harbourbridge schema -source=postgresql -source-profile="format=dump" -target=spanner < mydb.pg_dump
-```
-
-Similarly, you can use the tool for other source databases.
-
-```sh
-# Generate Spanner schema from a MySQL dump file
-mysqldump mydb > mydb.mysqldump
-harbourbridge schema -source=mysql < mydb.mysqldump
-```
-
-Generate Spanner schema by directly connecting to a DynamoDB database,
-```sh
-# Set the environment variables `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`
-# , `AWS_REGION` and then run,
-harbourbridge schema -source=dynamodb
-```
-
-Note: HarbourBridge accepts pg_dump/mysqldump's standard plain-text format, 
-but not archive or custom formats. More details on usage can be found in 
-[Example usage](#example-usage) section.
-
-HarbourBridge automatically determines the cloud project to use, and generates
-a new Spanner database name (prefixed with `{source}_` and today's date).
-Command-line flags can be used to explicitly set the Spanner instance or
-database name. See [Options](#options).
-
-**WARNING: Please check that permissions for the Spanner instance used by
-HarbourBridge are appropriate. Spanner manages access control at the database
-level, and the database created by HarbourBridge will inherit default
-permissions from the instance. All data written by HarbourBridge is visible to
-anyone who can access the created database.**
-
-As it processes the data, HarbourBridge reports on progress, provides
-stats on the schema and data conversion steps, and an overall assessment of the
-quality of the conversion. It also generates a schema file, report file and
-a session file (and a bad-data file if data was dropped). See
-[Files Generated by HarbourBridge](#files-generated-by-harbourbridge). Details
-of how source database's schema is mapped to Spanner can be found in the
-[Schema Conversion](#schema-conversion) section.
-
-This tool is part of the Cloud Spanner Ecosystem, a community contributed and
-supported open source repository. Please [report
-issues](https://github.com/cloudspannerecosystem/harbourbridge/issues) and send
-pull requests. See the [HarbourBridge
-Whitepaper](https://github.com/cloudspannerecosystem/harbourbridge/blob/master/whitepaper.md)
-for a discussion of our plans for the tool.
-
-Note that the HarbourBridge tool is not an officially supported Google product
-and is not officially supported as part of the Cloud Spanner product.
-
-## Quickstart Guide
-
-### Before you begin
-
-Complete the steps described in
-[Set up](https://cloud.google.com/spanner/docs/getting-started/set-up), which
-covers creating and setting a default Google Cloud project, enabling billing,
-enabling the Cloud Spanner API, and setting up OAuth 2.0 to get authentication
-credentials to use the Cloud Spanner API.
-
-In particular, ensure that you run
-
-```sh
-gcloud auth application-default login
-```
-
-to set up your local development environment with authentication credentials.
-
-Set the GCLOUD_PROJECT environment variable to your Google Cloud project ID:
-
-```sh
-export GCLOUD_PROJECT=my-project-id
-```
-
-If you do not already have a Cloud Spanner instance, or you want to use a
-separate instance specifically for running HarbourBridge, then create a Cloud
-Spanner instance by following the "Create an instance" instructions on the
-[Quickstart using the console](https://cloud.google.com/spanner/docs/quickstart-console)
-guide. HarbourBridge will create a database for you, but it will not create a
-Spanner instance.
-
-Install Go ([download](https://golang.org/doc/install)) on your development
-machine if it is not already installed, configure the GOPATH environment
-variable if it is not already configured, and
-[test your installation](https://golang.org/doc/install#testing).
-
-### Installing HarbourBridge
-
-Download the tool to your machine and install it.
-
-```sh
-GO111MODULE=on go get github.com/cloudspannerecosystem/harbourbridge
-```
-
-The tool should now be installed as `$GOPATH/bin/harbourbridge`.
-See the [Troubleshooting Guide](#troubleshooting-guide) if you run
-into any issues.
-
-### Running HarbourBridge
-
-To use the tool on a PostgreSQL database called mydb, run
-
-```sh
-pg_dump mydb > mydb.pg_dump
-$GOPATH/bin/harbourbridge schema -source=postgresql < mydb.pg_dump
-```
-
-To use the tool on a MySQL database called mydb, run
-
-```sh
-mysqldump mydb > mydb.mysqldump
-$GOPATH/bin/harbourbridge schema -source=mysql < mydb.mysqldump
-```
-
-To use the tool on a DynamoDB database, run
-
-```sh
-$GOPATH/bin/harbourbridge schema -source=dynamodb
-```
-More details on running harbourbridge can be found in [Example usage](#example-usage) section.
-
-This command will use the cloud project specified by the `GCLOUD_PROJECT`
-environment variable, automatically determine the Cloud Spanner instance
-associated with this project, convert the source schema to a Spanner schema 
-(For MySQL/Postgres) or infer a schema from the DynamoDB instance, create a
-new Cloud Spanner database with this schema, and finally, populate this new
-database with the data from the source database. 
-If the project has multiple instances, then list of available instances 
-will be shown and you will have to pick one of the available instances and 
-set the `--instance` flag in target-profile. The new Cloud Spanner database
-will have a name of the form `{SOURCE}_{DATE}_{RANDOM}`, where`{SOURCE}` is
-the source specified,`{DATE}`is today's date, and`{RANDOM}` is a random
-suffix for uniqueness.
-
-See the [Troubleshooting Guide](#troubleshooting-guide) for help on debugging
-issues.
-
-HarbourBridge also [generates several files](#files-generated-by-harbourbridge)
-when it runs: a schema file, a report file (with detailed analysis of the
-conversion), a session file and a bad data file (if any data was dropped).
-
-### Sample Dump Files
-
-If you don't have ready access to a PostgreSQL or MySQL database, some example
-dump files can be found [here](examples). The files
-[cart.pg_dump](examples/cart.pg_dump) and
-[cart.mysqldump](examples/cart.mysqldump) contain pg_dump and mysqldump output
-for a very basic shopping cart application (just two tables, one for products
-and one for user carts). The files [singers.pg_dump](examples/singers.pg_dump)
-and [singers.mysqldump](examples/singers.mysqldump) contain pg_dump and
-mysqldump output for a version of the [Cloud Spanner
-singers](https://cloud.google.com/spanner/docs/schema-and-data-model#creating_a_table)
-example. To use HarbourBridge on cart.pg_dump, download the file locally and
-run
-
-```
-$GOPATH/bin/harbourbridge schema -source=postgresql < cart.pg_dump
-```
-
-### Verifying Results
-
-Once the tool has completed, you can verify the new database and its content
-using the Google Cloud Console. Go to the [Cloud Spanner Instances
-page](https://console.cloud.google.com/spanner/instances), select your Spanner
-instance, and then find the database created by HarbourBridge and select
-it. This will list the tables created by HarbourBridge. Select a table, and take
-a look at its schema and data. Next, go to the query page, and try
-some SQL statements. For example
-
-```
-SELECT COUNT(*) from mytable
-```
-
-to check the number of rows in table `mytable`.
-
-### Next Steps
-
-The tables created by HarbourBridge provide a starting point for evaluation of
-Spanner. While they preserve much of the core structure of your PostgreSQL/MySQL
-schema and data, many key features have been dropped, including functions, 
-sequences, procedures,triggers, and views. For DynamoDB, the conversion from 
-schemaless to schema is focused on the use-case where customers use DynamoDB in 
-a consistent, structured way with a fairly well defined set of columns and types.
-
-As a result, the out-of-the-box performance you get from these tables could be
-slower than what you get from PostgreSQL/MySQL/DynamoDB. 
-
-To improve performance, also consider using [Interleaved
-Tables](https://cloud.google.com/spanner/docs/schema-and-data-model#creating-interleaved-tables)
-to tune performance.
-
-View HarbourBridge as a base set of functionality for Spanner evalution that can
-be readily expanded. Consider forking and modifying the codebase to add the
-functionality you need. Please [file
-issues](https://github.com/cloudspannerecosystem/harbourbridge/issues) and send
-PRs for fixes and new functionality. See our backlog of [open
-issues](https://github.com/cloudspannerecosystem/harbourbridge/issues). Our
-plans and aspirations for developing HarbourBridge further are outlined in the
-[HarbourBridge
-Whitepaper](https://github.com/cloudspannerecosystem/harbourbridge/blob/master/whitepaper.md).
-
-You can also change the way HarbourBridge behaves by directly editing the
-pg_dump/mysqldump output. For example, suppose you want to try out different 
-primary keys for a table. First run pg_dump/mysqldump and save the output to 
-a file. Then modify (or add) the relevant 
-`ALTER TABLE ... ADD CONSTRAINT ... PRIMARY KEY ...` statement in the 
-pg_dump/mysqldump output file so that the primary keys match what you need. 
-Then run HarbourBridge on the modified pg_dump/mysqldump output.
-
-## Files Generated by HarbourBridge
-
-HarbourBridge generates several files as it runs:
-
-- Schema file (ending in `schema.txt`): contains the generated Spanner
-  schema, interspersed with comments that cross-reference to the relevant
-  PostgreSQL/MySQL schema definitions.
-
-- Session file (ending in `session.json`): contains all schema and data
-  conversion state endcoded as JSON. It is basically a snapshot of the session.
-
-- Report file (ending in `report.txt`): contains a detailed analysis of the
-  PostgreSQL/MySQL to Spanner migration, including table-by-table stats and an
-  analysis of PostgreSQL/MySQL types that don't cleanly map onto Spanner types. 
-  Note that PostgreSQL/MySQL types that don't have a corresponding Spanner type 
-  are mapped to STRING(MAX).
-
-- Bad data file (ending in `dropped.txt`): contains details of data
-  that could not be converted and written to Spanner, including sample
-  bad-data rows. If there is no bad-data, this file is not written (and we
-  delete any existing file with the same name from a previous run).
-
-By default, these files are prefixed by the name of the Spanner database (with a
-dot separator). The file prefix can be overridden using the `-prefix`
-[option](#options).
-
-## Command line flags
-
-`-source` Required flag. Specifies the source database. Supported source
-databases are _'postgres'_, _'mysql'_ and _'dynamodb'_.
-
-`-target` Optional flag. Specifies the target database. Defaults to _'spanner'_ 
-, which is the only supported target database today.
-
-`-prefix` Specifies a file prefix for the report, schema, and bad-data files
-written by the tool. If no file prefix is specified, the name of the Spanner
-database (plus a '.') is used.
-
-`-v` or `-verbose` Specifies verbose mode. This will cause HarbourBridge to
-output detailed messages about the conversion.
-
-`-skip-foreign-keys` Controls whether we add foreign key constraints after
-data migration is complete. This flag cannot be used with schema-only mode,
-and does not affect the generation of foreign key statements during schema
-processing i.e. foreign key constraints will still appear in the generated 
-Spanner DDL files.
-
-`-session` Specifies a session file that contains all schema and data 
-conversion state endcoded as JSON.
-
-### Source profile (`-source-profile`)
-
-HarbourBridge accepts the following options for --source-profile,
-specified as "key1=value1,key2=value,..." pairs:
-
-`-file` Specifies the full path of the file to use for reading source database
-schema and/or data. This can also point to a file stored in GCS e.g., 
-gcs://<bucket_name>/<file_path>. This flag is optional, and file can also be
-piped to stdin, if available locally.
-
-`-format` Specifies the format of the file. This flag is also optional, and
-defaults to `dump`. This may be extended in future to support other formats
-such as `csv`, `avro` etc.
-
-### Target profile (`-target-profile`)
-
-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.
-
-`-instance` Specifies the Spanner instance to use. The new database will be
-created in this instance. If not specified, the tool automatically determines an
-appropriate instance using gcloud.
-
-## Example Usage
-
-Details on HarbourBridge example usage can be found here: 
-- [PostgreSQL example usage](sources/postgres/README.md#example-postgresql-usage)
-- [MySQL example usage](sources/mysql/README.md#example-mysql-usage)
-- [DynamoDB example usage](sources/dynamodb/README.md#example-dynamodb-usage)
-
-
-## Schema Conversion
-
-Details on HarbourBridge schema conversion can be found here:
-- [PostgreSQL schema conversion](sources/postgres/README.md#schema-conversion)
-- [MySQL schema conversion](sources/mysql/README.md#schema-conversion)
-- [DynamoDB schema conversion](sources/dynamodb/README.md#schema-conversion)
-
-## Data Conversion
-
-HarbourBridge converts PostgreSQL/MySQL/DynamoDB data to Spanner data based on 
-the Spanner schema it constructs. Conversion for most data types is fairly 
-straightforward, but several types deserve discussion. Note that HarbourBridge 
-is not intended for databases larger than about 100GB. Details on HarbourBridge 
-data conversion can be found here:
-- [PostgreSQL data conversion](sources/postgres/README.md#data-conversion)
-- [MySQL data conversion](sources/mysql/README.md#data-conversion)
-- [DynamoDB data conversion](sources/dynamodb/README.md#data-conversion)
-
-## Troubleshooting Guide
-
-HarbourBridge is written using the Go module system, and so it must be
-installed in [module-aware mode](https://golang.org/cmd/go/#hdr-Module_support).
-This can be achived by setting the environment variable `GO111MODULE` to `on`
-before calling "go get" e.g.
-
-```sh
-GO111MODULE=on go get github.com/cloudspannerecosystem/harbourbridge
-```
-
-As an alternative to "go get", you can make a copy of the HarbourBridge
-codebase from the github repository and use "go run":
-
-```sh
-git clone https://github.com/cloudspannerecosystem/harbourbridge
-cd harbourbridge
-pg_dump mydb > mydb.pg_dump
-go run github.com/cloudspannerecosystem/harbourbridge schema -source=postgresql < mydb.pg_dump
-```
-
-This workflow also allows you to modify or customize the HarbourBridge codebase.
-
-The following steps can help diagnose common issues encountered while running
-HarbourBridge.
-
-### 1. Verify driver configuration
-
-First, check that the driver is correctly configured to connect to your
-database. Driver configuration varies depending on the driver used.
-
-#### 1.1 pg_dump
-
-If you are using pg_dump (-driver=pg_dump), check that pg_dump is
-correctly configured to connect to your PostgreSQL
-database. Note that pg_dump uses the same options as psql to connect to your
-database. See the [psql](https://www.postgresql.org/docs/9.3/app-psql.html) and
-[pg_dump](https://www.postgresql.org/docs/9.3/app-pgdump.html) documentation.
-
-Access to a PostgreSQL database is typically configured using the
-_PGHOST_, _PGPORT_, _PGUSER_, _PGDATABASE_ environment variables,
-which are standard across PostgreSQL utilities.
-
-It is also possible to configure access via pg_dump's command-line options
-`--host`, `--port`, and `--username`.
-
-#### 1.2 Direct access to PostgreSQL
-
-In this case, HarbourBridge connects directly to the PostgreSQL
-database to retrieve table schema and data. Set the `-driver=postgres`
-and provide the environment variables `PGHOST`, `PGPORT`, `PGUSER`, 
-`PGDATABASE`. Password can be specified either in the `PGPASSWORD`
-environment variable or provided at the password prompt.
-
-#### 1.3 mysqldump
-
-If you are using mysqldump (-driver=mysqldump), check that mysqldump is
-correctly configured to connect to your MySQL
-database via the command-line options `--host`, `--port`, and `--user`.
-Note that mysqldump uses the same options as mysql to connect to your
-database. See the [mysql](https://dev.mysql.com/doc/refman/8.0/en/mysql-commands.html)
-and [mysqldump](https://dev.mysql.com/doc/refman/8.0/en/mysqldump.html) documentation.
-
-#### 1.4 Direct access to MySQL
-
-In this case, HarbourBridge connects directly to the MySQL
-database to retrieve table schema and data. Set the `-driver=mysql`
-and provide the environment variables `MYSQLHOST`, `MYSQLPORT`, 
-`MYSQLUSER`, `MYSQLDATABASE`. Password can be specified either 
-in the `MYSQLPWD` environment variable or provided at the password 
-prompt.
-
-
-#### 1.5 Direct access to DynamoDB
-
-In this case, HarbourBridge connects directly to the DynamoDB
-database to retrieve table schema and data. Set the `-driver=dynamodb`
-and provide the environment variables `AWS_ACCESS_KEY_ID`, 
-`AWS_SECRET_ACCESS_KEY`, `AWS_REGION`. If you use a custom endpoint for 
-dynamodb, you can specify that using the environment variable 
-`DYNAMODB_ENDPOINT_OVERRIDE`.
-
-### 2. Verify dump output
-
-Next, verify that pg_dump/mysqldump is generating plain-text output. If your 
-database is small, try running
-
-```sh
-{ pg_dump/mysqldump } > file
-```
-
-and look at the output file. It should be a plain-text file containing SQL
-commands. If your database is large, consider just dumping the schema via the
-`--schema-only` for pg_dump and `--no-data` for mysqldump command-line option.
-
-pg_dump/mysqldump can export data in a variety of formats, but HarbourBridge 
-only accepts `plain` format (aka plain-text). See the
-[pg_dump documentation](https://www.postgresql.org/docs/9.3/app-pgdump.html) and
-[mysqldump documentation](https://dev.mysql.com/doc/refman/8.0/en/mysqldump.html) 
-for details about formats.
-
-### 3. Debugging HarbourBridge
-
-The HarbourBridge tool can fail for a number of reasons.
-
-#### 3.1 No space left on device
-
-HarbourBridge needs to read the pg_dump/mysqldump output twice, once to build 
-a schema and once for data ingestion. When pg_dump/mysqldump output is directly
-piped to HarbourBridge, `stdin` is not seekable, and so we write the output to 
-a temporary file. That temporary file is created via Go's ioutil.TempFile. 
-On many systems, this creates a file in `/tmp`, which is sometimes configured 
-with minimal space. A simple workaround is to separately run pg_dump/mysqldump 
-and write its output to a file in a directory with sufficient space. For example,
-if the current working directory has space, then:
-
-```sh
-{ pg_dump/mysqldump } > tmpfile
-harbourbridge < tmpfile
-```
-
-Make sure you cleanup the tmpfile after HarbourBridge has been run. Another
-option is to set the location of Go's TempFile e.g. by setting the `TMPDIR`
-environment variable.
-
-#### 3.2 Unparsable dump output
-
-HarbourBridge uses the [pg_query_go](https://github.com/pganalyze/pg_query_go)
-library for parsing pg_dump and [pingcap parser](https://github.com/pingcap/parser) 
-for parsing mysqldump. It is possible that the pg_dump/mysqldump output is 
-corrupted or uses features that aren't parseable. Parsing errors should 
-generate an error message of the form `Error parsing last 54321 line(s) of input`.
-
-#### 3.2 Credentials problems
-
-HarbourBridge uses standard Google Cloud credential mechanisms for accessing
-Cloud Spanner. If this is mis-configured, you may see errors containing
-"unauthenticated", or "cannot fetch token", or "could not find default
-credentials". You might need to run `gcloud auth application-default login`. 
-See the [Before you begin](#before-you-begin) section for details.
-
-#### 3.4 Can't create database
-
-In this case, the error message printed by the tool should help identify the
-cause. It could be an API permissions issue. For example, the Cloud Spanner API
-may not be appropriately configured. See [Before you begin](#before-you-begin)
-section for details. Alternatively, you have have hit the limit on the number of
-databases per instances (currently 100). This can occur if you re-run the
-HarbourBridge tool many times, since each run creates a new database. In this
-case you'll need to [delete some
-databases](https://cloud.google.com/spanner/docs/getting-started/go/#delete_the_database).
-
-### 4. Database-Specific Issues
-
-The schema, report, and bad-data files [generated by
-HarbourBridge](#files-generated-by-harbourbridge) contain detailed information
-about the schema and data conversion process, including issues and problems
-encountered.
-
-### 5. Reporting Issues
-
-If you are having problems with HarbourBridge, please [submit an
-issue](https://github.com/cloudspannerecosystem/harbourbridge/issues).
diff --git a/cmd/README.mysql.md b/cmd/README.mysql.md
deleted file mode 100644
index 5e9394c..0000000
--- a/cmd/README.mysql.md
+++ /dev/null
@@ -1,268 +0,0 @@
-# HarbourBridge: Turnkey MySQL-to-Spanner Evaluation
-
-HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
-using data from an existing PostgreSQL or MySQL database. This README provides
-details of the tool's MySQL capabilities. For general HarbourBridge information
-see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-turnkey-spanner-evaluation).
-
-## Example MySQL Usage
-
-The following examples assume `harbourbridge` has been added to your PATH
-environment variable.
-
-HarbourBridge can either be used with mysqldump or it can be run directly
-on a MySQL database (via go's database/sql package).
-
-### Using HarbourBridge with mysqldump
-
-To use HarbourBridge on a MySQL database called mydb using mysqldump, run:
-
-```sh
-mysqldump mydb | harbourbridge schema -source=mysql
-```
-
-The tool can also be applied to an existing mysqldump file:
-
-```sh
-harbourbridge schema -source=mysql < my_mysqldump_file
-```
-
-To specify a particular Spanner instance to use during data migration, run:
-
-```sh
-harbourbridge data -source=mysql -target-profile="instance=my-spanner-instance" < my_mysqldump_file
-```
-
-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 -source=mysql -target-profile="dbname=my-spanner-database-name" < my_mysqldump_file
-```
-
-HarbourBridge generates a report file, a schema file, and a bad-data file (if
-there are bad-data rows). You can control where these files are written by
-specifying a file prefix. For example,
-
-```sh
-harbourbridge schema -prefix=mydb. -source=mysql < my_mysqldump_file
-```
-
-will write files `mydb.report.txt`, `mydb.schema.txt`, and
-`mydb.dropped.txt`. The prefix can also be a directory. For example,
-
-```sh
-harbourbridge schema -prefix=~/spanner-eval-mydb/ -source=mysql < my_mysqldump_file
-```
-
-would write the files into the directory `~/spanner-eval-mydb/`. Note
-that HarbourBridge will not create directories as it writes these files.
-
-### Directly connecting to a MySQL database
-
-To use the tool directly on a MySQL database called mydb, run
-
-```sh
-harbourbridge schema -source=mysql
-```
-
-It is assumed that _MYSQLHOST_, _MYSQLPORT_, _MYSQLUSER_, _MYSQLDATABASE_ environment
-variables are set. Password can be specified either in the _MYSQLPWD_ environment
-variable or provided at the password prompt.
-
-Note that all of the options described in the previous section on read from a
-mysqldump file can also be used when directly connecting to a mysql database.
-
-## Schema Conversion
-
-The HarbourBridge tool maps MySQL types to Spanner types as follows:
-
-| MySQL Type                                        | Spanner Type    | Notes                           |
-| ------------------------------------------------- | --------------- | ------------------------------- |
-| `BOOL`, `BOOLEAN`,<br/>`TINYINT(1)`               | `BOOL`          |                                 |
-| `BIGINT`                                          | `INT64`         |                                 |
-| `BINARY`, `VARBINARY`                             | `BYTES(MAX)`    |                                 |
-| `BLOB`, `MEDIUMBLOB`,<br/>`TINYBLOB`, `LONGBLOB`  | `BYTES(MAX)`    |                                 |
-| `BIT`                                             | `BYTES(MAX)`    |                                 |
-| `CHAR`                                            | `STRING(1)`     | CHAR defaults to length 1       |
-| `CHAR(N)`                                         | `STRING(N)`     | c                               |
-| `DATE`                                            | `DATE`          |                                 |
-| `DATETIME`                                        | `TIMESTAMP`     | t                               |
-| `DECIMAL`, `NUMERIC`                              | `NUMERIC`       | p                               |
-| `DOUBLE`                                          | `FLOAT64`       |                                 |
-| `ENUM`                                            | `STRING(MAX)`   |                                 |
-| `FLOAT`                                           | `FLOAT64`       | s                               |
-| `INTEGER`, `MEDIUMINT`,<br/>`TINYINT`, `SMALLINT` | `INT64`         | s                               |
-| `JSON`                                            | `STRING(MAX)`   |                                 |
-| `SET`                                             | `ARRAY<STRING>` | SET only supports string values |
-| `TEXT`, `MEDIUMTEXT`,<br/>`TINYTEXT`, `LONGTEXT`  | `STRING(MAX)`   |                                 |
-| `TIMESTAMP`                                       | `TIMESTAMP`     |                                 |
-| `VARCHAR`                                         | `STRING(MAX)`   |                                 |
-| `VARCHAR(N)`                                      | `STRING(N)`     | c                               |
-
-Spanner does not support `spatial` datatypes of MySQL. Along with `spatial`
-datatypes, all other types map to `STRING(MAX)`. Some of the mappings in this
-table represent potential changes of precision (marked p), differences in
-treatment of timezones (marked t), differences in treatment of fixed-length
-character types (marked c), and changes in storage size (marked s). We discuss
-these, as well as other limits and notes on schema conversion, in the following
-sections.
-
-### `DECIMAL` and `NUMERIC`
-
-[Spanner's NUMERIC
-type](https://cloud.google.com/spanner/docs/data-types#decimal_type) can store
-up to 29 digits before the decimal point and up to 9 after the decimal point.
-MySQL's NUMERIC type can potentially support higher precision than this, so
-please verify that Spanner's NUMERIC support meets your application needs.  Note
-that in MySQL, NUMERIC is implemented as DECIMAL, so the remarks about DECIMAL
-apply equally to NUMERIC.
-
-### `TIMESTAMP` and `DATETIME`
-
-MySQL has two timestamp types: `TIMESTAMP` and `DATETIME`. Both provide
-microsecond resolution, but neither actually stores a timezone with the data.
-The key difference between the two types is that MySQL converts `TIMESTAMP` values
-from the current time zone to UTC for storage, and back from UTC to the current time
-zone for retrieval. This does not occur for `DATETIME` and data is returned without a
-timezone. For `TIMESTAMP`, timezone can be set by time zone offset parameter.
-
-Spanner has a single timestamp type. Data is stored as UTC (there is no separate
-timezone) Spanner client libraries convert timestamps to UTC before sending them
-to Spanner. Data is always returned as UTC. Spanner's timestamp type is
-essentially the same as `TIMESTAMP`, except that there is no analog of
-MySQL's timezone offset parameter.
-
-In other words, mapping MySQL `DATETIME` to `TIMESTAMP` is fairly
-straightforward, but care should be taken with MySQL `DATETIME` data
-because Spanner clients will not drop the timezone.
-
-### `CHAR(n)` and `VARCHAR(n)`
-
-The semantics of fixed-length character types differ between MySQL and
-Spanner. The `CHAR(n)` type in MySQL is right-padded with spaces. If a string
-value smaller than the limit is stored, spaces will be added to pad it out to
-the specified length. If a string longer than the specified length is stored,
-and the extra characters are all spaces, then it will be silently
-truncated. Moreover, trailing spaces are ignored when comparing two values. In
-constrast, Spanner does not give special treatment to spaces, and the specified
-length simply represents the maximum length that can be stored. This is close to
-the semantics of MySQL's `VARCHAR(n)`. However there are some minor
-differences. For example, even `VARCHAR(n)` has some special treatment of
-spaces: string with trailing spaces in excess of the column length are truncated
-prior to insertion and a warning is generated.
-
-### `SET`
-
-MySQL `SET` is a string object that can hold muliple values, each of which must be
-chosen from a list of permitted values specified when the table is created. `SET`
-is being mapped to Spanner type `ARRAY<STRING>`. Validation of `SET` element values
-will be dropped in Spanner. Thus for production use, validation needs to be done
-in the application.
-
-### `Spatial datatype`
-
-MySQL spatial datatypes are used to represent geographic feature.
-It includes `GEOMETRY`, `POINT`, `LINESTRING`, `POLYGON`, `MULTIPOINT`, `MULTIPOLYGON`
-and `GEOMETRYCOLLECTION` datatypes. Spanner does not support spatial data types.
-This datatype are currently mapped to standard `STRING` Spanner datatype.
-
-### Storage Use
-
-The tool maps several MySQL types to Spanner types that use more storage.
-For example, `SMALLINT` is a two-byte integer, but it maps to Spanner's `INT64`,
-an eight-byte integer.
-
-### Primary Keys
-
-Spanner requires primary keys for all tables. MySQL recommends the use of
-primary keys for all tables, but does not enforce this. When converting a table
-without a primary key, HarbourBridge will create a new primary key of type
-INT64. By default, the name of the new column is `synth_id`. If there is already
-a column with that name, then a variation is used to avoid collisions.
-
-### NOT NULL Constraints
-
-The tool preserves `NOT NULL` constraints. Note that Spanner does not require
-primary key columns to be `NOT NULL`. However, in MySQL, a primary key is a
-combination of `NOT NULL` and `UNIQUE`, and so primary key columns from
-MySQL will be mapped to Spanner columns that are both primary keys and `NOT NULL`.
-
-### Foreign Keys
-
-The tool maps MySQL foreign key constraints into Spanner foreign key constraints, and
-preserves constraint names where possible. Since Spanner doesn't support `ON DELETE`
-and `ON UPDATE` actions, we drop them.
-
-### Default Values
-
-Spanner does not currently support default values. We drop these
-MySQL features during conversion.
-
-### Secondary Indexes
-
-The tool maps MySQL secondary indexes to Spanner secondary indexes, and preserves
-constraint names where possible. Note that Spanner requires index key constraint
-names to be globally unique (within a database), but in MySQL they only have to be
-unique for a table, so we add a uniqueness suffix to a name if needed. The tool also
-maps `UNIQUE` constraint into `UNIQUE` secondary index. Note that due to limitations of our
-mysqldump parser, we are not able to handle key column ordering (i.e. ASC/DESC) in
-mysqldump files. All key columns in mysqldump files will be treated as ASC.
-
-### Other MySQL features
-
-MySQL has many other features we haven't discussed, including functions,
-sequences, procedures, triggers, (non-primary) indexes and views. The tool does
-not support these and the relevant statements are dropped during schema
-conversion.
-
-See
-[Migrating from MySQL to Cloud Spanner](https://cloud.google.com/solutions/migrating-mysql-to-spanner)
-for a general discussion of MySQL to Spanner migration issues.
-HarbourBridge follows most of the recommendations in that guide. The main
-difference is that we map a few more types to `STRING(MAX)`.
-
-## Data Conversion
-
-### Timestamps and Timezones
-
-As noted earlier when discussing [schema conversion of
-TIMESTAMP](#timestamp), there are some subtle differences in how timestamps are
-handled in MySQL and Spanner.
-
-During data conversion, MySQL `TIMESTAMP` values are converted to UTC and
-stored in Spanner. The conversion proceeds as follows. If the value has a
-timezone offset, that timezone is respected during the conversion to UTC. If the value
-does not have a timezone offset, then we look for any `set timezone` statements in the
-mysqldump output and use the timezone offset specified. Otherwise, we use '+00:00' timezone offset (UTC).
-
-### Strings, character set support and UTF-8
-
-Spanner requires that `STRING` values be UTF-8 encoded. All Spanner functions
-and operators that act on `STRING` values operate on Unicode characters rather
-than bytes. Since we map many MySQL types (including `TEXT` and `CHAR`
-types) to Spanner's `STRING` type, HarbourBridge is effectively a UTF-8 based
-tool.
-
-Note that the tool itself does not do any encoding/decoding or UTF-8 checks: it
-passes through data from mysqldump to Spanner. Internally, we use Go's string
-type, which supports UTF-8.
-
-### Spatial datatypes support
-
-As noted earlier when discussing [schema conversion of
-Spatial datatype](#spatial-datatype), Spanner does not support spatial datatypes and are
-mapped to `STRING(MAX)` Spanner type. Data conversion for spatial datatypes is different
-for both `mysql` and `mysqldump` driver.
-
-- MySQL information schema approach (`-driver=mysql`) : Data from MySQL is fetched using
-  'ST_AsText(g)' function which converts a value in internal geometry format to its WKT(Well-Known Text)
-  representation and returns the string result. This value will be stored as `STRING` in Spanner.
-- MySQL dump approach (`-driver=mysqldump`) : Mysqldump will have the internal geometry data in
-  binary format. It cannot be converted to WKT format and there is no proper method for mysqldump
-  generation of spatial datatypes also. Thus, this value will just be fetched as a `TEXT` type and
-  converted to Spanner type `STRING`.
-
-Note that mysql information schema approach would be the recommended approach for data conversion of
-spatial datatypes. For production use, you must store this data using standard data types, and implement
-any searching/filtering logic in the application layer.
diff --git a/cmd/README.postgres.md b/cmd/README.postgres.md
deleted file mode 100644
index ab78960..0000000
--- a/cmd/README.postgres.md
+++ /dev/null
@@ -1,257 +0,0 @@
-# HarbourBridge: Turnkey PostgreSQL-to-Spanner Evaluation
-
-HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
-using data from an existing PostgreSQL or MySQL database. This README provides
-details of the tool's PostgreSQL capabilities. For general HarbourBridge information
-see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-turnkey-spanner-evaluation).
-
-## Example PostgreSQL Usage
-
-The following examples assume `harbourbridge` has been added to your PATH
-environment variable.
-
-HarbourBridge can either be used with pg_dump or it can be run directly
-on a PostgreSQL database (via go's database/sql package).
-
-### Using HarbourBridge with pg_dump
-
-To use HarbourBridge on a PostgreSQL database called mydb using pg_dump output,run:
-
-```sh
-pg_dump mydb | harbourbridge schema -source=postgresql
-```
-
-You can use one of `postgresql`, `postgres`, or `pg` to specify PostgreSQL as
-the source database.
-
-The tool can also be applied to an existing pg_dump file:
-
-```sh
-harbourbridge schema -source=postgres < my_pg_dump_file
-```
-
-To specify a particular Spanner instance to use during data migration, run:
-
-```sh
-harbourbridge data -source=postgres -target-profile="instance=my-spanner-instance" < my_pg_dump_file
-```
-
-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 -source=postgres -target-profile="dbname=my-spanner-database-name,instance=my-spanner-instance" < my_pg_dump_file
-```
-
-HarbourBridge generates a report file, a schema file, and a bad-data file (if
-there are bad-data rows during data migration). You can control where these
-files are written by specifying a file prefix. For example,
-
-```sh
-harbourbridge schema -prefix=mydb. -source=pg < my_pg_dump_file
-```
-
-will write files `mydb.report.txt`, `mydb.schema.txt`, and
-`mydb.dropped.txt`. The prefix can also be a directory. For example,
-
-```sh
-harbourbridge schema -prefix=~/spanner-eval-mydb/ -source=pg < my_pg_dump_file
-```
-
-would write the files into the directory `~/spanner-eval-mydb/`. Note
-that HarbourBridge will not create directories as it writes these files.
-
-### Directly connecting to a PostgreSQL database
-
-To use the tool directly on a PostgresSQL database called mydb, run
-
-```sh
-harbourbridge schema -source=postgres
-```
-
-It is assumed that _PGHOST_, _PGPORT_, _PGUSER_, _PGDATABASE_ environment
-variables are set. Password can be specified either in the _PGPASSWORD_
-environment variable or provided at the password prompt.
-
-Note that all of the options described in the previous section when specifying
-a dump file can also be used when directly connecting to a PostgreSQL database.
-
-## Schema Conversion
-
-The HarbourBridge tool maps PostgreSQL types to Spanner types as follows:
-
-| PostgreSQL Type    | Spanner Type           | Notes                                     |
-| ------------------ | ---------------------- | ----------------------------------------- |
-| `BOOL`             | `BOOL`                 |                                           |
-| `BIGINT`           | `INT64`                |                                           |
-| `BIGSERIAL`        | `INT64`                | a                                         |
-| `BYTEA`            | `BYTES(MAX)`           |                                           |
-| `CHAR`             | `STRING(1)`            | CHAR defaults to length 1                 |
-| `CHAR(N)`          | `STRING(N)`            | c                                         |
-| `DATE`             | `DATE`                 |                                           |
-| `DOUBLE PRECISION` | `FLOAT64`              |                                           |
-| `INTEGER`          | `INT64`                | s                                         |
-| `NUMERIC`          | `NUMERIC`              | p                                         |
-| `REAL`             | `FLOAT64`              | s                                         |
-| `SERIAL`           | `INT64`                | a, s                                      |
-| `SMALLINT`         | `INT64`                | s                                         |
-| `TEXT`             | `STRING(MAX)`          |                                           |
-| `TIMESTAMP`        | `TIMESTAMP`            | t                                         |
-| `TIMESTAMPTZ`      | `TIMESTAMP`            |                                           |
-| `VARCHAR`          | `STRING(MAX)`          |                                           |
-| `VARCHAR(N)`       | `STRING(N)`            | c                                         |
-| `ARRAY(`pgtype`)`  | `ARRAY(`spannertype`)` | if scalar type pgtype maps to spannertype |
-
-All other types map to `STRING(MAX)`. Some of the mappings in this table
-represent potential changes of precision (marked p), dropped autoincrement
-functionality (marked a), differences in treatment of timezones (marked t),
-differences in treatment of fixed-length character types (marked c), and changes
-in storage size (marked s). We discuss these, as well as other limits and notes
-on schema conversion, in the following sections.
-
-### `NUMERIC`
-
-[Spanner's NUMERIC
-type](https://cloud.google.com/spanner/docs/data-types#decimal_type) can store
-up to 29 digits before the decimal point and up to 9 after the decimal point.
-PostgreSQL's NUMERIC type can potentially support higher precision that this, so
-please verify that Spanner's NUMERIC support meets your application needs.
-
-### `BIGSERIAL` and `SERIAL`
-
-Spanner does not support autoincrementing types, so these both map to `INT64`
-and the autoincrementing functionality is dropped.
-
-### `TIMESTAMP`
-
-PosgreSQL has two timestamp types: `TIMESTAMP` and `TIMESTAMPTZ`. Both have an 8
-byte data representation and provide microsecond resolution, but neither
-actually stores a timezone with the data. The key difference between the two
-types is how string literals are converted to timestamps and queries return
-data. For `TIMESTAMP`, all timezone information is dropped, and data is returned
-without a timezone. For `TIMESTAMPTZ`, string literals are converted to UTC,
-using the literal's timezone if it is specified, or the PostgreSQL's timezone
-paramater if not. When data is printed stored data (in UTC) is converted to the
-timezone from the timezone parameter
-
-Spanner has a single timestamp type. Data is stored as UTC (there is no separate
-timezone) Spanner client libraries convert timestamps to UTC before sending them
-to Spanner. Data is always returned as UTC. Spanner's timestamp type is
-essentially the same as `TIMESTAMPTZ`, except that there is no analog of
-PostgreSQL's timezone parameter.
-
-In other words, mapping PostgreSQL `TIMESTAMPTZ` to `TIMESTAMP` is fairly
-straightforward, but care should be taken with PostgreSQL `TIMESTAMP` data
-because Spanner clients will not drop the timezone.
-
-### `CHAR(n)` and `VARCHAR(n)`
-
-The semantics of fixed-length character types differ between PostgreSQL and
-Spanner. The `CHAR(n)` type in PostgreSQL is padded with spaces. If a string
-value smaller than the limit is stored, spaces will be added to pad it out to
-the specified length. If a string longer than the specified length is stored,
-and the extra characters are all spaces, then it will be silently
-truncated. Moreover, trailing spaces are ignored when comparing two values. In
-constrast, Spanner does not give special treatment to spaces, and the specified
-length simply represents the maximum length that can be stored. This is close to
-the semantics of PostgreSQL's `VARCHAR(n)`. However there are some minor
-differences. For example, even `VARCHAR(n)` has some special treatment of
-spaces: strings longer than the specified length are silently truncated if the
-extra characters are all spaces.
-
-### Storage Use
-
-The tool maps several PostgreSQL types to Spanner types that use more storage.
-For example, `SMALLINT` is a two-byte integer, but it maps to Spanner's `INT64`,
-an eight-byte integer. This additional storage could be significant for large
-arrays.
-
-### Arrays
-
-Spanner does not support multi-dimensional arrays. So while `TEXT[4]` maps to
-`ARRAY<STRING(MAX)>` and `REAL ARRAY` maps to `ARRAY<FLOAT64>`, `TEXT[][]` maps
-to `STRING(MAX)`.
-
-Also note that PosgreSQL supports array limits, but the PostgreSQL
-implementation ignores them. Spanner does not support array size limits, but
-since they have no effect anyway, the tool just drops them.
-
-### Primary Keys
-
-Spanner requires primary keys for all tables. PostgreSQL recommends the use of
-primary keys for all tables, but does not enforce this. When converting a table
-without a primary key, HarbourBridge will create a new primary key of type
-INT64. By default, the name of the new column is `synth_id`. If there is already
-a column with that name, then a variation is used to avoid collisions.
-
-### NOT NULL Constraints
-
-The tool preserves `NOT NULL` constraints. Note that Spanner does not require
-primary key columns to be `NOT NULL`. However, in PostgreSQL, a primary key is a
-combination of `NOT NULL` and `UNIQUE`, and so primary key columns from
-PostgreSQL will be mapped to Spanner columns that are both primary keys and `NOT NULL`.
-
-### Foreign Keys
-
-The tool maps PostgreSQL foreign key constraints into Spanner foreign key constraints, and
-preserves constraint names where possible. Note that Spanner requires foreign key
-constraint names to be globally unique (within a database), but in postgres they only
-have to be unique for a table, so we add a uniqueness suffix to a name if needed.
-Spanner doesn't support `ON DELETE` and `ON UPDATE` actions, so we drop these.
-
-### Default Values
-
-Spanner does not currently support default values. We drop these
-PostgreSQL features during conversion.
-
-### Secondary Indexes
-
-The tool maps PostgresSQL secondary indexes to Spanner secondary indexes, preserving
-constraint names where possible. The tool also maps PostgreSQL `UNIQUE` constraints to
-Spanner `UNIQUE` secondary indexes. Check [here](https://cloud.google.com/spanner/docs/migrating-postgres-spanner#indexes)
-for more details.
-
-### Other PostgreSQL features
-
-PostgreSQL has many other features we haven't discussed, including functions,
-sequences, procedures, triggers, (non-primary) indexes and views. The tool does
-not support these and the relevant statements are dropped during schema
-conversion.
-
-See
-[Migrating from PostgreSQL to Cloud Spanner](https://cloud.google.com/spanner/docs/migrating-postgres-spanner)
-for a general discussion of PostgreSQL to Spanner migration issues.
-HarbourBridge follows most of the recommendations in that guide. The main
-difference is that we map a few more types to `STRING(MAX)`.
-
-## Data Conversion
-
-### Timestamps and Timezones
-
-As noted earlier when discussing [schema conversion of
-TIMESTAMP](#timestamp), there are some subtle differences in how timestamps are
-handled in PostgreSQL and Spanner.
-
-During data conversion, PostgreSQL `TIMESTAMPTZ` values are converted to UTC and
-stored in Spanner. The conversion proceeds as follows. If the value has a
-timezone, that timezone is respected during the conversion to UTC. If the value
-does not have a timezone, then we look for any `set timezone` statements in the
-pg_dump output and use the timezone specified. Otherwise, we use the `TZ`
-environment variable as the timezone, and failing that, we use the local system
-timezone default (as determined by Go).
-
-In constrast, conversion of PostgreSQL `TIMESTAMP` values proceeds by ignoring
-any timezone information and just treating the value as UTC and storing it in
-Spanner.
-
-### Strings, character set support and UTF-8
-
-Spanner requires that `STRING` values be UTF-8 encoded. All Spanner functions
-and operators that act on `STRING` values operate on Unicode characters rather
-than bytes. Since we map many PostgreSQL types (including `TEXT` and `CHAR`
-types) to Spanner's `STRING` type, HarbourBridge is effectively a UTF-8 based
-tool.
-
-Note that the tool itself does not do any encoding/decoding or UTF-8 checks: it
-passes through data from pg_dump to Spanner. Internally, we use Go's string
-type, which supports UTF-8.
diff --git a/cmd/cmd.go b/cmd/cmd.go
index 84bba51..ebd4a7c 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -81,7 +81,7 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 
 	// We pass an empty string to the sqlConnectionStr parameter as this is the legacy codepath,
 	// which reads the environment variables and constructs the string later on.
-	bw, err := conversion.DataConv(driver, "", ioHelper, client, conv, dataOnly)
+	bw, err := conversion.DataConv(driver, "", ioHelper, client, conv, dataOnly, schemaSampleSize)
 	if err != nil {
 		return fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 	}
diff --git a/cmd/data.go b/cmd/data.go
index 7fb1d7b..0b40afd 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -41,7 +41,7 @@ func (cmd *DataCmd) Usage() string {
 Migrate data from source db to target db. Source db dump file can be specified
 by either file param in source-profile or piped to stdin. Connection profile
 for source databases in direct connect mode can be specified by setting
-appropriate environment variables. The data flags are:
+appropriate params in source-profile. The data flags are:
 `, path.Base(os.Args[0]))
 }
 
@@ -129,7 +129,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(driverName, getSQLConnectionStr(sourceProfile), &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(driverName, getSQLConnectionStr(sourceProfile), &ioHelper, client, conv, true, getSchemaSampleSize(sourceProfile))
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
diff --git a/cmd/schema.go b/cmd/schema.go
index 219a9a2..d1e9d44 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -38,8 +38,8 @@ func (cmd *SchemaCmd) Usage() string {
 
 Convert schema for source db specified by source and source-profile. Source db
 dump file can be specified by either file param in source-profile or piped to
-stdin. Connection profile for source databases in direct connect mode can be
-specified by setting appropriate environment variables. The schema flags are:
+stdin. Connection profile for source database in direct connect mode can be
+specified by setting appropriate params in source-profile. The schema flags are:
 `, path.Base(os.Args[0]))
 }
 
diff --git a/cmd/eval.go b/cmd/schema_and_data.go
similarity index 84%
rename from cmd/eval.go
rename to cmd/schema_and_data.go
index 95726ce..47657e3 100644
--- a/cmd/eval.go
+++ b/cmd/schema_and_data.go
@@ -13,8 +13,8 @@ import (
 	"github.com/google/subcommands"
 )
 
-// EvalCmd struct with flags.
-type EvalCmd struct {
+// SchemaAndDataCmd struct with flags.
+type SchemaAndDataCmd struct {
 	source          string
 	sourceProfile   string
 	target          string
@@ -24,28 +24,28 @@ type EvalCmd struct {
 }
 
 // Name returns the name of operation.
-func (cmd *EvalCmd) Name() string {
-	return "eval"
+func (cmd *SchemaAndDataCmd) Name() string {
+	return "schema-and-data"
 }
 
 // Synopsis returns summary of operation.
-func (cmd *EvalCmd) Synopsis() string {
-	return "evaluate schema and data migration from source db to target db"
+func (cmd *SchemaAndDataCmd) Synopsis() string {
+	return "schema and data migration from source db to target db in schema-and-data"
 }
 
 // Usage returns usage info of the command.
-func (cmd *EvalCmd) Usage() string {
-	return fmt.Sprintf(`%v eval -source=[source] -target-profile="instance=my-instance"...
+func (cmd *SchemaAndDataCmd) Usage() string {
+	return fmt.Sprintf(`%v schema-and-data -source=[source] -target-profile="instance=my-instance"...
 
-Evaluate schema and data migration from source db to target db. Source db dump
+Migrate schema and data from source db to target db in schema-and-data. Source db dump
 file can be specified by either file param in source-profile or piped to stdin.
 Connection profile for source databases in direct connect mode can be specified
-by setting appropriate environment variables. The eval flags are:
+by setting appropriate params in source-profile. The schema-and-data flags are:
 `, path.Base(os.Args[0]))
 }
 
 // SetFlags sets the flags.
-func (cmd *EvalCmd) SetFlags(f *flag.FlagSet) {
+func (cmd *SchemaAndDataCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.source, "source", "", "Flag for specifying source DB, (e.g., `PostgreSQL`, `MySQL`, `DynamoDB`)")
 	f.StringVar(&cmd.sourceProfile, "source-profile", "", "Flag for specifying connection profile for source database e.g., \"file=<path>,format=dump\"")
 	f.StringVar(&cmd.target, "target", "Spanner", "Specifies the target DB, defaults to Spanner (accepted values: `Spanner`)")
@@ -54,7 +54,7 @@ func (cmd *EvalCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
 }
 
-func (cmd *EvalCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
+func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
 	var err error
 	defer func() {
 		if err != nil {
@@ -98,9 +98,9 @@ func (cmd *EvalCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 	}
 
 	sqlConnectionStr := getSQLConnectionStr(sourceProfile)
-
+	schemaSampleSize := getSchemaSampleSize(sourceProfile)
 	var conv *internal.Conv
-	conv, err = conversion.SchemaConv(driverName, sqlConnectionStr, targetDb, &ioHelper, getSchemaSampleSize(sourceProfile))
+	conv, err = conversion.SchemaConv(driverName, sqlConnectionStr, targetDb, &ioHelper, schemaSampleSize)
 	if err != nil {
 		panic(err)
 	}
@@ -134,7 +134,7 @@ func (cmd *EvalCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(driverName, sqlConnectionStr, &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(driverName, sqlConnectionStr, &ioHelper, client, conv, true, schemaSampleSize)
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
diff --git a/cmd/source_profile.go b/cmd/source_profile.go
index 92312fa..84d09cd 100644
--- a/cmd/source_profile.go
+++ b/cmd/source_profile.go
@@ -173,10 +173,10 @@ type SourceProfileConnectionDynamoDB struct {
 
 func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfileConnectionDynamoDB, error) {
 	dydb := SourceProfileConnectionDynamoDB{}
-	if schemaSampleSize, ok := params["schemaSampleSize"]; ok {
+	if schemaSampleSize, ok := params["schema-sample-size"]; ok {
 		schemaSampleSizeInt, err := strconv.Atoi(schemaSampleSize)
 		if err != nil {
-			return dydb, fmt.Errorf("could not parse schemaSampleSize = %v as a valid int64", schemaSampleSize)
+			return dydb, fmt.Errorf("could not parse schema-sample-size = %v as a valid int64", schemaSampleSize)
 		}
 		dydb.schemaSampleSize = int64(schemaSampleSizeInt)
 	}
@@ -184,16 +184,16 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 	// Unlike postgres and mysql, there may not be deprecation of env variables, hence it
 	// is better to override env variables optionally via source profile params.
 	var ok bool
-	if dydb.awsAccessKeyID, ok = params["awsAccessKeyID"]; ok {
+	if dydb.awsAccessKeyID, ok = params["aws-access-key-id"]; ok {
 		os.Setenv("AWS_ACCESS_KEY_ID", dydb.awsAccessKeyID)
 	}
-	if dydb.awsSecretAccessKey, ok = params["awsSecretAccessKey"]; ok {
+	if dydb.awsSecretAccessKey, ok = params["aws-secret-access-key"]; ok {
 		os.Setenv("AWS_SECRET_ACCESS_KEY", dydb.awsAccessKeyID)
 	}
-	if dydb.awsRegion, ok = params["awsRegion"]; ok {
+	if dydb.awsRegion, ok = params["aws-region"]; ok {
 		os.Setenv("AWS_REGION", dydb.awsAccessKeyID)
 	}
-	if dydb.dydbEndpoint, ok = params["dydbEndpoint"]; ok {
+	if dydb.dydbEndpoint, ok = params["dydb-endpoint"]; ok {
 		os.Setenv("DYNAMODB_ENDPOINT_OVERRIDE", dydb.awsAccessKeyID)
 	}
 	return dydb, nil
diff --git a/cmd/target_profile.go b/cmd/target_profile.go
index c0c2645..e5e5b09 100644
--- a/cmd/target_profile.go
+++ b/cmd/target_profile.go
@@ -53,16 +53,16 @@ func (trg TargetProfile) ToLegacyTargetDb() string {
 				{
 					sp := conn.sp
 					if len(sp.dialect) > 0 && strings.ToLower(sp.dialect) == constants.DIALECT_POSTGRESQL {
-						return constants.TARGET_EXPERIMENTAL_POSTGRES
+						return constants.TargetExperimentalPostgres
 					}
-					return constants.TARGET_SPANNER
+					return constants.TargetSpanner
 				}
 			default:
-				return constants.TARGET_SPANNER
+				return constants.TargetSpanner
 			}
 		}
 	default:
-		return constants.TARGET_SPANNER
+		return constants.TargetSpanner
 	}
 }
 
diff --git a/common/constants/constants.go b/common/constants/constants.go
index 9631b91..2768c8f 100644
--- a/common/constants/constants.go
+++ b/common/constants/constants.go
@@ -21,8 +21,8 @@ const (
 	DYNAMODB string = "dynamodb"
 
 	// Target db for which schema is being generated.
-	TARGET_SPANNER               string = "spanner"
-	TARGET_EXPERIMENTAL_POSTGRES string = "experimental_postgres"
+	TargetSpanner              string = "spanner"
+	TargetExperimentalPostgres string = "experimental_postgres"
 
 	// Supported dialects for Cloud Spanner database.
 	DIALECT_POSTGRESQL string = "postgresql"
diff --git a/conversion/conversion.go b/conversion/conversion.go
index d92008c..703ffac 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -20,7 +20,6 @@
 // 			key public type definitions next (although often it makes sense to put them next to public functions that use them)
 // 			then public functions (and relevant type definitions)
 // 			and helper functions and other non-public definitions last (generally in order of importance)
-
 package conversion
 
 import (
@@ -48,8 +47,6 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	dydb "github.com/aws/aws-sdk-go/service/dynamodb"
-	_ "github.com/go-sql-driver/mysql"
-	_ "github.com/lib/pq"
 	"golang.org/x/crypto/ssh/terminal"
 	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
@@ -76,6 +73,10 @@ var (
 	MaxWorkers = 20
 )
 
+// 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.
@@ -83,23 +84,22 @@ var (
 // When using source-profile, the sqlConnectionStr is constructed from the input params.
 func SchemaConv(driver, sqlConnectionStr, targetDb string, ioHelper *IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
 	switch driver {
-	case constants.POSTGRES, constants.MYSQL:
-		return schemaFromSQL(driver, sqlConnectionStr, targetDb)
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB:
+		return schemaFromDatabase(driver, sqlConnectionStr, targetDb, schemaSampleSize)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		return schemaFromDump(driver, targetDb, ioHelper)
-	case constants.DYNAMODB:
-		return schemaFromDynamoDB(schemaSampleSize)
 	default:
 		return nil, fmt.Errorf("schema conversion for driver %s not supported", driver)
 	}
 }
 
+// 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 is constructed from the input params.
-func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
+// When using source-profile, the sqlConnectionStr and schemaSampleSize are constructed from the input params.
+func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool, schemaSampleSize int64) (*spanner.BatchWriter, error) {
 	config := spanner.BatchWriterConfig{
 		BytesLimit: 100 * 1000 * 1000,
 		WriteLimit: 40,
@@ -107,26 +107,36 @@ func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.C
 		Verbose:    internal.Verbose(),
 	}
 	switch driver {
-	case constants.POSTGRES, constants.MYSQL:
-		return dataFromSQL(driver, sqlConnectionStr, config, client, conv)
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB:
+		return dataFromDatabase(driver, sqlConnectionStr, config, client, conv, schemaSampleSize)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		if conv.SpSchema.CheckInterleaved() {
-			return nil, fmt.Errorf("HarbourBridge does not currently support data conversion from dump files\nif the schema contains interleaved tables. Suggest using direct access to source database\ni.e. using drivers postgres and mysql.")
+			return nil, fmt.Errorf("harbourBridge does not currently support data conversion from dump files\nif the schema contains interleaved tables. Suggest using direct access to source database\ni.e. using drivers postgres and mysql")
 		}
 		return dataFromDump(driver, config, ioHelper, client, conv, dataOnly)
-	case constants.DYNAMODB:
-		return dataFromDynamoDB(config, client, conv)
 	default:
 		return nil, fmt.Errorf("data conversion for driver %s not supported", driver)
 	}
 }
 
-func generateSQLConnectionStr(driver string) (string, error) {
+func connectionConfig(driver string, sqlConnectionStr string) (interface{}, error) {
 	switch driver {
 	case constants.POSTGRES:
-		return generatePGSQLConnectionStr()
+		// If empty, this is called as part of the legacy mode witih global CLI flags.
+		// When using source-profile mode is used, the sqlConnectionStr is already populated.
+		if sqlConnectionStr == "" {
+			return generatePGSQLConnectionStr()
+		}
+		return sqlConnectionStr, nil
 	case constants.MYSQL:
-		return generateMYSQLConnectionStr()
+		// If empty, this is called as part of the legacy mode witih global CLI flags.
+		// When using source-profile mode is used, the sqlConnectionStr is already populated.
+		if sqlConnectionStr == "" {
+			return generateMYSQLConnectionStr()
+		}
+		return sqlConnectionStr, nil
+	case constants.DYNAMODB:
+		return getDynamoDBClientConfig()
 	default:
 		return "", fmt.Errorf("driver %s not supported", driver)
 	}
@@ -183,55 +193,22 @@ func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
 	return ""
 }
 
-// TODO: Pass around cmd.SourceProfile instead of sqlConnectionStr.
-// Doing that requires refactoring since that would introduce a circular dependency between
-// conversion.go and cmd/source_profile.go.
-func schemaFromSQL(driver, sqlConnectionStr, targetDb string) (*internal.Conv, error) {
-	// If empty, this is called as part of the legacy mode witih global CLI flags.
-	// When using source-profile mode is used, the sqlConnectionStr is already populated.
-	if sqlConnectionStr == "" {
-		var err error
-		sqlConnectionStr, err = generateSQLConnectionStr(driver)
-		if err != nil {
-			return nil, err
-		}
-	}
-	sourceDB, err := sql.Open(driver, sqlConnectionStr)
-	if err != nil {
-		return nil, err
-	}
+func schemaFromDatabase(driver, sqlConnectionStr, targetDb string, schemaSampleSize int64) (*internal.Conv, error) {
 	conv := internal.MakeConv()
 	conv.TargetDb = targetDb
-	conv.SrcDbName = getDbNameFromSQLConnectionStr(driver, sqlConnectionStr)
-	err = ProcessInfoSchema(driver, conv, sourceDB)
+	infoSchema, err := GetInfoSchema(driver, sqlConnectionStr, schemaSampleSize)
 	if err != nil {
-		return nil, err
+		return conv, err
 	}
-	return conv, nil
+	return conv, common.ProcessSchema(conv, infoSchema)
 }
 
-func dataFromSQL(driver, sqlConnectionStr string, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
-	// TODO: Refactor to avoid redundant calls to sqlConnectionStr and
-	// Open in schemaFromSQL and dataFromSQL. Also refactor to
-	// share code with dataFromPgDump. Use single transaction for
-	// reading schema and data from source db to get consistent
-	// dump.
-	if sqlConnectionStr == "" {
-		// If empty, this is called as part of the old command line workflow.
-		var err error
-		sqlConnectionStr, err = generateSQLConnectionStr(driver)
-		if err != nil {
-			return nil, err
-		}
-	}
-	sourceDB, err := sql.Open(driver, sqlConnectionStr)
-	if err != nil {
-		return nil, err
-	}
-	err = SetRowStats(driver, conv, sourceDB)
+func dataFromDatabase(driver, sqlConnectionStr string, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv, schemaSampleSize int64) (*spanner.BatchWriter, error) {
+	infoSchema, err := GetInfoSchema(driver, sqlConnectionStr, schemaSampleSize)
 	if err != nil {
 		return nil, err
 	}
+	common.SetRowStats(conv, infoSchema)
 	totalRows := conv.Rows()
 	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
 	rows := int64(0)
@@ -250,66 +227,21 @@ func dataFromSQL(driver, sqlConnectionStr string, config spanner.BatchWriterConf
 		func(table string, cols []string, vals []interface{}) {
 			writer.AddRow(table, cols, vals)
 		})
-	err = ProcessSQLData(driver, conv, sourceDB)
-	if err != nil {
-		return nil, err
-	}
+	common.ProcessData(conv, infoSchema)
 	writer.Flush()
 	return writer, nil
 }
 
-func getDynamoDBClientConfig() *aws.Config {
+func getDynamoDBClientConfig() (*aws.Config, error) {
 	cfg := aws.Config{}
 	endpointOverride := os.Getenv("DYNAMODB_ENDPOINT_OVERRIDE")
 	if endpointOverride != "" {
 		cfg.Endpoint = aws.String(endpointOverride)
 	}
-	return &cfg
-}
-
-func schemaFromDynamoDB(sampleSize int64) (*internal.Conv, error) {
-	conv := internal.MakeConv()
-	mySession := session.Must(session.NewSession())
-	dydbClient := dydb.New(mySession, getDynamoDBClientConfig())
-	err := dynamodb.ProcessSchema(conv, dydbClient, []string{}, sampleSize)
-	if err != nil {
-		return nil, err
-	}
-	return conv, nil
-}
-
-func dataFromDynamoDB(config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
-	mySession := session.Must(session.NewSession())
-	dydbClient := dydb.New(mySession, getDynamoDBClientConfig())
-	dynamodb.SetRowStats(conv, dydbClient)
-	totalRows := conv.Rows()
-	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
-
-	rows := int64(0)
-	config.Write = func(m []*sp.Mutation) error {
-		_, err := client.Apply(context.Background(), m)
-		if err != nil {
-			return err
-		}
-		atomic.AddInt64(&rows, int64(len(m)))
-		p.MaybeReport(atomic.LoadInt64(&rows))
-		return nil
-	}
-	writer := spanner.NewBatchWriter(config)
-	conv.SetDataMode()
-	conv.SetDataSink(
-		func(table string, cols []string, vals []interface{}) {
-			writer.AddRow(table, cols, vals)
-		})
-
-	err := dynamodb.ProcessData(conv, dydbClient)
-	if err != nil {
-		return nil, err
-	}
-	writer.Flush()
-	return writer, nil
+	return &cfg, nil
 }
 
+// IOStreams is a struct that contains the file descriptor for dumpFile.
 type IOStreams struct {
 	In, SeekableIn, Out *os.File
 	BytesRead           int64
@@ -530,7 +462,7 @@ func getSeekable(f *os.File) (*os.File, int64, error) {
 	if err != nil {
 		return nil, 0, fmt.Errorf("can't reset file offset: %w", err)
 	}
-	n, err := getSize(fcopy)
+	n, _ := getSize(fcopy)
 	return fcopy, n, nil
 }
 
@@ -566,11 +498,12 @@ func ValidateDDL(ctx context.Context, adminClient *database.DatabaseAdminClient,
 		return fmt.Errorf("can't fetch database ddl: %v", err)
 	}
 	if len(dbDdl.Statements) != 0 {
-		return fmt.Errorf("HarbourBridge supports writing to existing databases only if they have an empty schema")
+		return fmt.Errorf("harbourBridge supports writing to existing databases only if they have an empty schema")
 	}
 	return nil
 }
 
+// CreatesOrUpdatesDatabase updates an existing Spanner database or creates a new one if one does not exist.
 func CreateOrUpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
 	dbExists, err := VerifyDb(ctx, adminClient, dbURI)
 	if err != nil {
@@ -604,8 +537,8 @@ func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 	req := &adminpb.CreateDatabaseRequest{
 		Parent: fmt.Sprintf("projects/%s/instances/%s", project, instance),
 	}
-	if conv.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
-		// TARGET_EXPERIMENTAL_POSTGRES doesn't support:
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
+		// TargetExperimentalPostgres doesn't support:
 		// a) backticks around the database name, and
 		// b) DDL statements as part of a CreateDatabase operation (so schema
 		// must be set using a separate UpdateDatabase operation).
@@ -625,13 +558,14 @@ func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 	}
 	fmt.Fprintf(out, "Created database successfully.\n")
 
-	if conv.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
 		// Update schema separately for PG databases.
 		return UpdateDatabase(ctx, adminClient, dbURI, conv, out)
 	}
 	return nil
 }
 
+// UpdateDatabase updates an existing spanner database.
 func UpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
 	fmt.Fprintf(out, "Updating schema for %s with default permissions ... \n", dbURI)
 	// The schema we send to Spanner excludes comments (since Cloud
@@ -722,15 +656,15 @@ Recommended value is between 20-30.`)
 	// too many requests and get throttled due to network or hitting catalog memory limits.
 	// Ensure atmost `MaxWorkers` go routines run in parallel that each update the ddl with one foreign key statement.
 	for _, fkStmt := range fkStmts {
-		workerId := <-workers
-		go func(fkStmt string, workerId int) {
+		workerID := <-workers
+		go func(fkStmt string, workerID int) {
 			defer func() {
 				// Locking the progress reporting otherwise progress results displayed could be in random order.
 				progressMutex.Lock()
 				progress++
 				p.MaybeReport(progress)
 				progressMutex.Unlock()
-				workers <- workerId
+				workers <- workerID
 			}()
 			internal.VerbosePrintf("Submitting new FK create request: %s\n", fkStmt)
 			op, err := adminClient.UpdateDatabaseDdl(ctx, &adminpb.UpdateDatabaseDdlRequest{
@@ -748,7 +682,7 @@ Recommended value is between 20-30.`)
 				return
 			}
 			internal.VerbosePrintln("Updated schema with statement: " + fkStmt)
-		}(fkStmt, workerId)
+		}(fkStmt, workerID)
 	}
 	// Wait for all the goroutines to finish.
 	for i := 1; i <= MaxWorkers; i++ {
@@ -1005,21 +939,19 @@ func AnalyzeError(err error, URI string) error {
 	project, instance, _ := parseURI(URI)
 	e := strings.ToLower(err.Error())
 	if containsAny(e, []string{"unauthenticated", "cannot fetch token", "default credentials"}) {
-		return fmt.Errorf("%w.\n"+`
+		return fmt.Errorf("%w."+`
 Possible cause: credentials are mis-configured. Do you need to run
 
   gcloud auth application-default login
 
 or configure environment variable GOOGLE_APPLICATION_CREDENTIALS.
-See https://cloud.google.com/docs/authentication/getting-started.
-`, err)
+See https://cloud.google.com/docs/authentication/getting-started`, err)
 	}
 	if containsAny(e, []string{"instance not found"}) && instance != "" {
 		return fmt.Errorf("%w.\n"+`
 Possible cause: Spanner instance specified via instance option does not exist.
 Please check that '%s' is correct and that it is a valid Spanner
-instance for project %s.
-`, err, instance, project)
+instance for project %s`, err, instance, project)
 	}
 	return err
 }
@@ -1158,41 +1090,30 @@ func ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 	}
 }
 
-// ProcessInfoSchema invokes process infoschema function from a sql package based on driver selected.
-func ProcessInfoSchema(driver string, conv *internal.Conv, db *sql.DB) error {
-	switch driver {
-	case constants.MYSQL:
-		return common.ProcessInfoSchema(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
-	case constants.POSTGRES:
-		return common.ProcessInfoSchema(conv, db, postgres.InfoSchemaImpl{})
-	default:
-		return fmt.Errorf("schema conversion for driver %s not supported", driver)
-	}
-}
-
-// SetRowStats invokes SetRowStats function from a sql package based on driver selected.
-func SetRowStats(driver string, conv *internal.Conv, db *sql.DB) error {
-	switch driver {
-	case constants.MYSQL:
-		common.SetRowStats(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
-	case constants.POSTGRES:
-		common.SetRowStats(conv, db, postgres.InfoSchemaImpl{})
-	default:
-		return fmt.Errorf("could not set rows stats for '%s' driver", driver)
+func GetInfoSchema(driver, sqlConnectionStr string, schemaSampleSize int64) (common.InfoSchema, error) {
+	connectionConfig, err := connectionConfig(driver, sqlConnectionStr)
+	if err != nil {
+		return nil, err
 	}
-	return nil
-}
-
-// ProcessSQLData invokes ProcessSQLData function from a sql package based on driver selected.
-func ProcessSQLData(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
-	//TODO - move this logic into a factory within the sources dir
 	case constants.MYSQL:
-		common.ProcessSQLData(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
+		db, err := sql.Open(driver, connectionConfig.(string))
+		dbName := getDbNameFromSQLConnectionStr(driver, connectionConfig.(string))
+		if err != nil {
+			return nil, err
+		}
+		return mysql.InfoSchemaImpl{DbName: dbName, Db: db}, nil
 	case constants.POSTGRES:
-		common.ProcessSQLData(conv, db, postgres.InfoSchemaImpl{})
+		db, err := sql.Open(driver, connectionConfig.(string))
+		if err != nil {
+			return nil, err
+		}
+		return postgres.InfoSchemaImpl{Db: db}, nil
+	case constants.DYNAMODB:
+		mySession := session.Must(session.NewSession())
+		dydbClient := dydb.New(mySession, connectionConfig.(*aws.Config))
+		return dynamodb.InfoSchemaImpl{DynamoClient: dydbClient, SampleSize: schemaSampleSize}, nil
 	default:
-		return fmt.Errorf("data conversion for driver %s is not supported", driver)
+		return nil, fmt.Errorf("driver %s not supported", driver)
 	}
-	return nil
 }
diff --git a/go.mod b/go.mod
index 68e5515..073cf3f 100644
--- a/go.mod
+++ b/go.mod
@@ -8,11 +8,13 @@ require (
 	cloud.google.com/go/storage v1.17.0
 	github.com/DATA-DOG/go-sqlmock v1.4.1
 	github.com/aws/aws-sdk-go v1.34.5
+	github.com/form3tech-oss/jwt-go v3.2.5+incompatible // indirect
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/google/go-cmp v0.5.6
 	github.com/google/subcommands v1.2.0
 	github.com/gorilla/handlers v1.5.0
 	github.com/gorilla/mux v1.7.3
+	github.com/gorilla/websocket v1.4.1 // indirect
 	github.com/lib/pq v1.9.0
 	github.com/pganalyze/pg_query_go/v2 v2.0.5
 	//github.com/pingcap/parser v3.0.12+incompatible
@@ -24,6 +26,7 @@ require (
 	golang.org/x/tools v0.1.7 // indirect
 	google.golang.org/api v0.57.0
 	google.golang.org/genproto v0.0.0-20210921142501-181ce0d877f6
+	honnef.co/go/tools v0.2.1 // indirect
 )
 
 // cloud.google.com/go will upgrade grpc to v1.40.0
@@ -39,3 +42,7 @@ replace cloud.google.com/go => github.com/hengfengli/google-cloud-go v0.48.1-0.2
 replace cloud.google.com/go/spanner => github.com/hengfengli/google-cloud-go/spanner v1.1.1-0.20210903130221-989a19403c8c
 
 replace google.golang.org/genproto => github.com/hengfengli/go-genproto v0.0.0-20210914051132-d2ab1d87ef40
+
+// jwt-go, an indirect dependency imported through pingcap/tidb, is an unmaintained repo which has security issues.
+// Replacing it with the fork that tidb's latest master uses and has the security patch.
+replace github.com/dgrijalva/jwt-go => github.com/form3tech-oss/jwt-go v3.2.6-0.20210809144907-32ab6a8243d7+incompatible
diff --git a/go.sum b/go.sum
index d6644a2..f5c60c7 100644
--- a/go.sum
+++ b/go.sum
@@ -78,8 +78,6 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/denisenkom/go-mssqldb v0.0.0-20191124224453-732737034ffd/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
 github.com/dgraph-io/ristretto v0.0.1 h1:cJwdnj42uV8Jg4+KLrYovLiCgIfz9wtWm6E6KA+1tLs=
 github.com/dgraph-io/ristretto v0.0.1/go.mod h1:T40EBc7CJke8TkpiYfGGKAeFjSaxuFXhuXRyumBd6RE=
-github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
-github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
 github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
@@ -98,6 +96,10 @@ github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL
 github.com/fatih/structtag v1.2.0/go.mod h1:mBJUNpUnHmRKrKlQQlmCrh5PuhftFbNv8Ys4/aAZl94=
 github.com/felixge/httpsnoop v1.0.1 h1:lvB5Jl89CsZtGIWuTcDM1E/vkVs49/Ml7JJe07l8SPQ=
 github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
+github.com/form3tech-oss/jwt-go v3.2.5+incompatible h1:/l4kBbb4/vGSsdtB5nUe8L7B9mImVMaBPw9L/0TBHU8=
+github.com/form3tech-oss/jwt-go v3.2.5+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
+github.com/form3tech-oss/jwt-go v3.2.6-0.20210809144907-32ab6a8243d7+incompatible h1:0sWoh2EtO7UrQdNTAN+hnU3QXa4AoivplyPLLHkcrLk=
+github.com/form3tech-oss/jwt-go v3.2.6-0.20210809144907-32ab6a8243d7+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
@@ -209,8 +211,9 @@ github.com/gorilla/mux v1.7.3 h1:gnP5JzjVOuiZD07fKKToCAOjS0yOpj/qPETTXCCS6hw=
 github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
 github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.2.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
-github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
+github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM=
+github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4 h1:z53tR0945TRRQO/fLEVPI6SMv7ZflF0TEaTAoU7tOzg=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
@@ -456,6 +459,7 @@ github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q
 github.com/yookoala/realpath v1.0.0/go.mod h1:gJJMA9wuX7AcqLy1+ffPatSCySA1FQ2S8Ya9AIoYBpE=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 github.com/yuin/goldmark v1.4.0/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 go.etcd.io/bbolt v1.3.3 h1:MUGmc65QhB3pIlaQ5bB4LwqSj6GIonVJXpZiaKNyaKk=
@@ -553,6 +557,7 @@ golang.org/x/net v0.0.0-20200506145744-7e3656a0809f/go.mod h1:qpuaurCH72eLCgpAm/
 golang.org/x/net v0.0.0-20200513185701-a91f0712d120/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200520182314-0ba52f642ac2/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
@@ -571,6 +576,7 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -604,6 +610,7 @@ golang.org/x/sys v0.0.0-20200515095857-1151b9dac4a9/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200523222454-059865788121/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -661,6 +668,7 @@ golang.org/x/tools v0.0.0-20200501065659-ab2804fb9c9d/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20200512131952-2bc93b1c0c88/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200515010526-7d3b6ebf133d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200618134242-20370b0cb4b2/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.7 h1:6j8CgantCy3yc8JGBqkDLMKWqZ0RDU2g1HVgacojGWQ=
 golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo=
@@ -737,8 +745,9 @@ honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-honnef.co/go/tools v0.0.1-2020.1.4 h1:UoveltGrhghAA7ePc+e+QYDHXrBps2PqFZiHkGR/xK8=
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
+honnef.co/go/tools v0.2.1 h1:/EPr//+UMMXwMTkXvCCoaJDq8cpjMO80Ou+L4PDo2mY=
+honnef.co/go/tools v0.2.1/go.mod h1:lPVVZ2BS5TfnjLyizF7o7hv7j9/L+8cZY2hLyjP9cGY=
 k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
 sigs.k8s.io/yaml v1.1.0 h1:4A07+ZFc2wgJwo8YNlQpr1rVlgUDlxXHhPJciaPY5gs=
 sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
diff --git a/internal/convert.go b/internal/convert.go
index 356ca4b..acf0ed4 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -28,7 +28,6 @@ type Conv struct {
 	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.
-	SrcDbName      string                              // Name of the source database. Currently only used for mysql infoschema.
 	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.
diff --git a/internal/mapping.go b/internal/mapping.go
index cd28d27..3d3b72b 100644
--- a/internal/mapping.go
+++ b/internal/mapping.go
@@ -43,7 +43,7 @@ func GetSpannerTable(conv *Conv, srcTable string) (string, error) {
 	if sp, found := conv.ToSpanner[srcTable]; found {
 		return sp.Name, nil
 	}
-	spTable := getSpannerId(conv, srcTable)
+	spTable := getSpannerID(conv, srcTable)
 	if spTable != srcTable {
 		VerbosePrintf("Mapping source DB table %s to Spanner table %s\n", srcTable, spTable)
 	}
@@ -135,11 +135,11 @@ func GetSpannerCols(conv *Conv, srcTable string, srcCols []string) ([]string, er
 // (across the database). But in some source databases, such as PostgreSQL,
 // they only have to be unique for a table. Hence we must map each source
 // constraint name to a unique spanner constraint name.
-func ToSpannerForeignKey(conv *Conv, srcId string) string {
-	if srcId == "" {
+func ToSpannerForeignKey(conv *Conv, srcID string) string {
+	if srcID == "" {
 		return ""
 	}
-	return getSpannerId(conv, srcId)
+	return getSpannerID(conv, srcID)
 }
 
 // ToSpannerIndexName maps source index name to legal Spanner index name.
@@ -151,16 +151,16 @@ func ToSpannerForeignKey(conv *Conv, srcId string) string {
 // (across the database). But in some source databases, such as MySQL,
 // they only have to be unique for a table. Hence we must map each source
 // constraint name to a unique spanner constraint name.
-func ToSpannerIndexName(conv *Conv, srcId string) string {
-	return getSpannerId(conv, srcId)
+func ToSpannerIndexName(conv *Conv, srcID string) string {
+	return getSpannerID(conv, srcID)
 }
 
 // conv.UsedNames tracks Spanner names that have been used for table names, foreign key constraints
 // and indexes. We use this to ensure we generate unique names when
 // we map from source dbs to Spanner since Spanner requires all these names to be
 // distinct and should not differ only in case.
-func getSpannerId(conv *Conv, srcId string) string {
-	spKeyName, _ := FixName(srcId)
+func getSpannerID(conv *Conv, srcID string) string {
+	spKeyName, _ := FixName(srcID)
 	if _, found := conv.UsedNames[strings.ToLower(spKeyName)]; found {
 		// spKeyName has been used before.
 		// Add unique postfix: use number of keys so far.
@@ -226,7 +226,7 @@ func resolveTableRef(conv *Conv, tableRef string) (string, error) {
 			return t, nil
 		}
 	}
-	return "", fmt.Errorf("Can't resolve table %v", tableRef)
+	return "", fmt.Errorf("can't resolve table %v", tableRef)
 }
 
 func resolveColRefs(conv *Conv, tableRef string, colRefs []string) ([]string, error) {
@@ -245,7 +245,7 @@ func resolveColRefs(conv *Conv, tableRef string, colRefs []string) ([]string, er
 				return c, nil
 			}
 		}
-		return "", fmt.Errorf("Can't resolve column: table=%v, column=%v", tableRef, colRef)
+		return "", fmt.Errorf("can't resolve column: table=%v, column=%v", tableRef, colRef)
 	}
 	var cols []string
 	for _, colRef := range colRefs {
diff --git a/internal/report.go b/internal/report.go
index 5658186..c44c1fb 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -97,6 +97,7 @@ type tableReportBody struct {
 	Lines   []string
 }
 
+// AnalyzeTables generates table reports for all processed tables.
 func AnalyzeTables(conv *Conv, badWrites map[string]int64) (r []tableReport) {
 	// Process tables in alphabetical order. This ensures that tables
 	// appear in alphabetical order in report.txt.
@@ -119,7 +120,7 @@ func buildTableReport(conv *Conv, srcTable string, badWrites map[string]int64) t
 	if err != nil || !ok1 || !ok2 {
 		m := "bad source-DB-to-Spanner table mapping or Spanner schema"
 		conv.Unexpected("report: " + m)
-		tr.Body = []tableReportBody{tableReportBody{Heading: "Internal error: " + m}}
+		tr.Body = []tableReportBody{{Heading: "Internal error: " + m}}
 		return tr
 	}
 	issues, cols, warnings := analyzeCols(conv, srcTable, spTable)
@@ -247,7 +248,7 @@ func fillRowStats(conv *Conv, srcTable string, badWrites map[string]int64, tr *t
 	tr.badRows = badConvRows + badRowWrites
 }
 
-// Provides a description and severity for each schema issue.
+// IssueDB provides a description and severity for each schema issue.
 // Note on batch: for some issues, we'd like to report just the first instance
 // in a table and suppress other instances i.e. adding more instances
 // of the issue in the same table has little value and could be very noisy.
@@ -379,6 +380,7 @@ func rateConversion(rows, badRows, cols, warnings int64, missingPKey, summary bo
 	return rate
 }
 
+// GenerateSummary creates a summarized version of a tableReport.
 func GenerateSummary(conv *Conv, r []tableReport, badWrites map[string]int64) string {
 	cols := int64(0)
 	warnings := int64(0)
@@ -407,6 +409,7 @@ func GenerateSummary(conv *Conv, r []tableReport, badWrites map[string]int64) st
 	return rateConversion(rows, badRows, cols, warnings, missingPKey, true, conv.SchemaMode())
 }
 
+// IgnoredStatements creates a list of statements to ignore.
 func IgnoredStatements(conv *Conv) (l []string) {
 	for s := range conv.Stats.Statement {
 		switch s {
diff --git a/main.go b/main.go
index 64a7162..ffcfb8f 100644
--- a/main.go
+++ b/main.go
@@ -49,7 +49,7 @@ var (
 	sessionJSON      string
 	webapi           bool
 	dumpFilePath     string
-	targetDb         = constants.TARGET_SPANNER
+	targetDb         = constants.TargetSpanner
 )
 
 func setupGlobalFlags() {
@@ -66,7 +66,7 @@ func setupGlobalFlags() {
 	flag.StringVar(&sessionJSON, "session", "", "session: specifies the file we restore session state from (used in data-only to provide schema and data mapping)")
 	flag.BoolVar(&webapi, "web", false, "web: run the web interface (experimental)")
 	flag.StringVar(&dumpFilePath, "dump-file", "", "dump-file: location of dump file to process")
-	flag.StringVar(&targetDb, "target-db", constants.TARGET_SPANNER, "target-db: Specifies the target DB. Defaults to spanner")
+	flag.StringVar(&targetDb, "target-db", constants.TargetSpanner, "target-db: Specifies the target DB. Defaults to spanner")
 }
 
 func didSetVerboseTwice() bool {
@@ -107,7 +107,7 @@ func main() {
 		subcommands.Register(subcommands.CommandsCommand(), "")
 		subcommands.Register(&cmd.SchemaCmd{}, "")
 		subcommands.Register(&cmd.DataCmd{}, "")
-		subcommands.Register(&cmd.EvalCmd{}, "")
+		subcommands.Register(&cmd.SchemaAndDataCmd{}, "")
 		flag.Parse()
 		os.Exit(int(subcommands.Execute(ctx)))
 	}
@@ -121,7 +121,7 @@ func main() {
 
 	// Note: the web interface does not use any commandline flags.
 	if webapi {
-		web.WebApp()
+		web.App()
 		return
 	}
 
@@ -140,11 +140,11 @@ func main() {
 		panic(fmt.Errorf("can't use both schema-only and skip-foreign-keys at once, foreign Key creation can only be skipped when data migration takes place"))
 	}
 
-	if targetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if targetDb == constants.TargetExperimentalPostgres {
 		if !(driverName == constants.PGDUMP || driverName == constants.POSTGRES) {
 			panic(fmt.Errorf("can only convert to experimental postgres when source %s or %s. (target-db: %s driver: %s)", constants.PGDUMP, constants.POSTGRES, targetDb, driverName))
 		}
-	} else if targetDb != constants.TARGET_SPANNER {
+	} else if targetDb != constants.TargetSpanner {
 		panic(fmt.Errorf("unkown target-db %s", targetDb))
 	}
 	fmt.Printf("Using driver (source DB): %s target-db: %s\n", driverName, targetDb)
diff --git a/schema/schema.go b/schema/schema.go
index 41a22b0..49f10b2 100644
--- a/schema/schema.go
+++ b/schema/schema.go
@@ -34,6 +34,7 @@ import (
 // Table represents a database table.
 type Table struct {
 	Name        string
+	Schema      string
 	ColNames    []string          // List of column names (for predictable iteration order e.g. printing).
 	ColDefs     map[string]Column // Details of columns.
 	PrimaryKeys []Key
diff --git a/sources/common/dbdump.go b/sources/common/dbdump.go
index 365014a..f207dc5 100644
--- a/sources/common/dbdump.go
+++ b/sources/common/dbdump.go
@@ -18,6 +18,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 )
 
+// DbDump common interface for database dump functions.
 type DbDump interface {
 	GetToDdl() ToDdl
 	ProcessDump(conv *internal.Conv, r *internal.Reader) error
diff --git a/sources/common/infoschema.go b/sources/common/infoschema.go
index fd0e174..320dcca 100644
--- a/sources/common/infoschema.go
+++ b/sources/common/infoschema.go
@@ -15,7 +15,6 @@
 package common
 
 import (
-	"database/sql"
 	"fmt"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -23,25 +22,27 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
+// InfoSchema contains database information.
 type InfoSchema interface {
 	GetToDdl() ToDdl
 	GetTableName(schema string, tableName string) string
-	GetTables(db *sql.DB) ([]SchemaAndName, error)
-	GetColumns(table SchemaAndName, db *sql.DB) (*sql.Rows, error) //TODO - merge this method and ProcessColumns for cleaner interface
-	ProcessColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string)
-	GetRowsFromTable(conv *internal.Conv, db *sql.DB, table SchemaAndName) (*sql.Rows, error)
-	GetRowCount(db *sql.DB, table SchemaAndName) (int64, error)
-	GetConstraints(conv *internal.Conv, db *sql.DB, table SchemaAndName) ([]string, map[string][]string, error)
-	GetForeignKeys(conv *internal.Conv, db *sql.DB, table SchemaAndName) (foreignKeys []schema.ForeignKey, err error)
-	GetIndexes(conv *internal.Conv, db *sql.DB, table SchemaAndName) ([]schema.Index, error)
-	ProcessDataRows(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, rows *sql.Rows)
+	GetTables() ([]SchemaAndName, error)
+	GetColumns(conv *internal.Conv, table SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error)
+	GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error)
+	GetRowCount(table SchemaAndName) (int64, error)
+	GetConstraints(conv *internal.Conv, table SchemaAndName) ([]string, map[string][]string, error)
+	GetForeignKeys(conv *internal.Conv, table SchemaAndName) (foreignKeys []schema.ForeignKey, err error)
+	GetIndexes(conv *internal.Conv, table SchemaAndName) ([]schema.Index, error)
+	ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error
 }
 
+// SchemaAndName contains the schema and name for a table
 type SchemaAndName struct {
 	Schema string
 	Name   string
 }
 
+// FkConstraint contains foreign key constraints
 type FkConstraint struct {
 	Name    string
 	Table   string
@@ -49,16 +50,16 @@ type FkConstraint struct {
 	Cols    []string
 }
 
-// ProcessInfoSchema performs schema conversion for source database
+// ProcessSchema performs schema conversion for source database
 // 'db'. Information schema tables are a broadly supported ANSI standard,
 // and we use them to obtain source database's schema information.
-func ProcessInfoSchema(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) error {
-	tables, err := infoSchema.GetTables(db)
+func ProcessSchema(conv *internal.Conv, infoSchema InfoSchema) error {
+	tables, err := infoSchema.GetTables()
 	if err != nil {
 		return err
 	}
 	for _, t := range tables {
-		if err := processTable(conv, db, t, infoSchema); err != nil {
+		if err := processTable(conv, t, infoSchema); err != nil {
 			return err
 		}
 	}
@@ -67,68 +68,39 @@ func ProcessInfoSchema(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) e
 	return nil
 }
 
-// ProcessSQLData performs data conversion for source database
+// ProcessData performs data conversion for source database
 // 'db'. For each table, we extract and convert the data to Spanner data
 // (based on the source and Spanner schemas), and write it to Spanner.
 // If we can't get/process data for a table, we skip that table and process
 // the remaining tables.
-//
-// Using database/sql library we pass *sql.RawBytes to rows.scan.
-// RawBytes is a byte slice and values can be easily converted to string.
-func ProcessSQLData(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) {
-	// TODO: refactor to use the set of tables computed by
-	// ProcessInfoSchema instead of computing them again.
-	tables, err := infoSchema.GetTables(db)
-	if err != nil {
-		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
-		return
-	}
-	for _, t := range tables {
-		srcTable := infoSchema.GetTableName(t.Schema, t.Name)
-		srcSchema, ok := conv.SrcSchema[srcTable]
-		if !ok {
+func ProcessData(conv *internal.Conv, infoSchema InfoSchema) {
+	for srcTable, srcSchema := range conv.SrcSchema {
+		spTable, err1 := internal.GetSpannerTable(conv, srcTable)
+		spCols, err2 := internal.GetSpannerCols(conv, srcTable, srcSchema.ColNames)
+		spSchema, ok := conv.SpSchema[spTable]
+		if err1 != nil || err2 != nil || !ok {
 			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't get schemas for table %s", srcTable))
-			continue
-		}
-		rows, err := infoSchema.GetRowsFromTable(conv, db, t)
-		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", t.Name, err))
-			continue
-		}
-		defer rows.Close()
-		srcCols, _ := rows.Columns()
-		spTable, err := internal.GetSpannerTable(conv, srcTable)
-		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Couldn't get spanner table : %s", err))
+			conv.Unexpected(fmt.Sprintf("Can't get cols and schemas for table %s: err1=%s, err2=%s, ok=%t",
+				srcTable, err1, err2, ok))
 			continue
 		}
-		spCols, err := internal.GetSpannerCols(conv, srcTable, srcCols)
+		err := infoSchema.ProcessData(conv, srcTable, srcSchema, spTable, spCols, spSchema)
 		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Couldn't get spanner columns for table %s : err = %s", t.Name, err))
-			continue
+			return
 		}
-		spSchema, ok := conv.SpSchema[spTable]
-		if !ok {
-			//TODO - check why Bad rows are not being added in above conditions
-			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't get schemas for table %s", srcTable))
-			continue
-		}
-		infoSchema.ProcessDataRows(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, rows)
 	}
 }
 
 // SetRowStats populates conv with the number of rows in each table.
-func SetRowStats(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) {
-	tables, err := infoSchema.GetTables(db)
+func SetRowStats(conv *internal.Conv, infoSchema InfoSchema) {
+	tables, err := infoSchema.GetTables()
 	if err != nil {
 		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
 		return
 	}
 	for _, t := range tables {
 		tableName := infoSchema.GetTableName(t.Schema, t.Name)
-		count, err := infoSchema.GetRowCount(db, t)
+		count, err := infoSchema.GetRowCount(t)
 		if err != nil {
 			conv.Unexpected(fmt.Sprintf("Couldn't get number of rows for table %s", tableName))
 			continue
@@ -137,26 +109,20 @@ func SetRowStats(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) {
 	}
 }
 
-func processTable(conv *internal.Conv, db *sql.DB, table SchemaAndName, infoSchema InfoSchema) error {
-	cols, err := infoSchema.GetColumns(table, db)
-	if err != nil {
-		return fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
-	}
-	defer cols.Close()
-
-	primaryKeys, constraints, err := infoSchema.GetConstraints(conv, db, table)
+func processTable(conv *internal.Conv, table SchemaAndName, infoSchema InfoSchema) error {
+	primaryKeys, constraints, err := infoSchema.GetConstraints(conv, table)
 	if err != nil {
 		return fmt.Errorf("couldn't get constraints for table %s.%s: %s", table.Schema, table.Name, err)
 	}
-	foreignKeys, err := infoSchema.GetForeignKeys(conv, db, table)
+	foreignKeys, err := infoSchema.GetForeignKeys(conv, table)
 	if err != nil {
 		return fmt.Errorf("couldn't get foreign key constraints for table %s.%s: %s", table.Schema, table.Name, err)
 	}
-	indexes, err := infoSchema.GetIndexes(conv, db, table)
+	indexes, err := infoSchema.GetIndexes(conv, table)
 	if err != nil {
 		return fmt.Errorf("couldn't get indexes for table %s.%s: %s", table.Schema, table.Name, err)
 	}
-	colDefs, colNames := infoSchema.ProcessColumns(conv, cols, constraints)
+	colDefs, colNames, err := infoSchema.GetColumns(conv, table, constraints, primaryKeys)
 	if err != nil {
 		return fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
 	}
@@ -167,6 +133,7 @@ func processTable(conv *internal.Conv, db *sql.DB, table SchemaAndName, infoSche
 	}
 	conv.SrcSchema[name] = schema.Table{
 		Name:        name,
+		Schema:      table.Schema,
 		ColNames:    colNames,
 		ColDefs:     colDefs,
 		PrimaryKeys: schemaPKeys,
diff --git a/sources/common/toddl.go b/sources/common/toddl.go
index fea23d5..e10bb43 100644
--- a/sources/common/toddl.go
+++ b/sources/common/toddl.go
@@ -39,7 +39,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
-// This interface is meant to be implemented by all sources. When support for a
+// ToDdl interface is meant to be implemented by all sources. When support for a
 // new target database is added, please add a new method here with the output
 // type expected. In case a particular source to target transoformation is not
 // supported, an error is to be returned by the corresponding method.
@@ -47,7 +47,7 @@ type ToDdl interface {
 	ToSpannerType(conv *internal.Conv, columnType schema.Type) (ddl.Type, []internal.SchemaIssue)
 }
 
-// schemaToSpannerDDL performs schema conversion from the source DB schema to
+// SchemaToSpannerDDL performs schema conversion from the source DB schema to
 // Spanner. It uses the source schema in conv.SrcSchema, and writes
 // the Spanner schema to conv.SpSchema.
 func SchemaToSpannerDDL(conv *internal.Conv, toddl ToDdl) error {
diff --git a/sources/common/utils.go b/sources/common/utils.go
index 43f9dd6..3a38a02 100644
--- a/sources/common/utils.go
+++ b/sources/common/utils.go
@@ -6,6 +6,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 )
 
+// ToNotNull returns true if a column is not nullable and false if it is.
 func ToNotNull(conv *internal.Conv, isNullable string) bool {
 	switch isNullable {
 	case "YES":
diff --git a/sources/dynamodb/README.md b/sources/dynamodb/README.md
index 5d076e9..76eba9f 100644
--- a/sources/dynamodb/README.md
+++ b/sources/dynamodb/README.md
@@ -1,54 +1,103 @@
-# HarbourBridge: Turnkey DynamoDB-to-Spanner Evaluation
+# HarbourBridge: DynamoDB-to-Spanner Evaluation and Migration
 
-HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
-using data from an existing PostgreSQL or MySQL or DynamoDB database. This
+HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation and migration,
+using data from an existing database. This
 README provides details of the tool's DynamoDB capabilities. For general
-HarbourBridge information see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-turnkey-spanner-evaluation).
+HarbourBridge information see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-spanner-evaluation-and-migration).
 
 ## Example DynamoDB Usage
 
-The following examples assume `harbourbridge` alias has been setup as
-following.
-
-```sh
-git clone https://github.com/cloudspannerecosystem/harbourbridge
-cd harbourbridge
-alias harbourbridge="go run github.com/cloudspannerecosystem/harbourbridge"
-```
-
 Before running HarbourBridge, make sure that you have
 [set up your AWS credentials/region](https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html)
-correctly (set the environment variables `AWS_ACCESS_KEY_ID`, 
-`AWS_SECRET_ACCESS_KEY`, `AWS_REGION`). HarbourBridge accesses your 
+correctly (set the environment variables `AWS_ACCESS_KEY_ID`,
+`AWS_SECRET_ACCESS_KEY`, `AWS_REGION`). HarbourBridge accesses your
 DynamoDB database via the aws go sdk package. If you use a custom endpoint
-for dynamodb, you can specify that using the environment variable 
+for dynamodb, you can specify that using the environment variable
 `DYNAMODB_ENDPOINT_OVERRIDE`.
 
+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.
+
+For example, run
+
+```sh
+export DYNAMODB_ENDPOINT_OVERRIDE=http://dynamodb.us-west-2.amazonaws.com
+harbourbridge schema -source=dynamodb 
+```
+
+Instead of setting the environment variables, you
+can also pass corresponding source profile connection parameters `aws-access-key-id`
+, `aws-secret-access-key`, `aws-region`. Custom endpoint can be specified using
+`dydb-endpoint` param.
+
+For example, to convert schema run
+
+```sh
+harbourbridge schema -source=dynamodb -source-profile="aws-access-key-id=<>,aws-secret-access-key=<>,aws-region=<>"
+```
+
+This will generate a session file with `session.json` suffix. This file contains
+schema mapping from source to destination. You will need to specify this file
+during data migration. You can also specify a particular Spanner instance to use
+during data migration.
+
+For example, run
+
+```sh
+harbourbridge data -session=mydb.session.json -source=dynamodb -source-profile="aws-access-key-id=<>,..." -target-profile="instance=my-spanner-instance"
+```
+
+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
-export DYNAMODB_ENDPOINT_OVERRIDE=http://dynamodb.us-west-2.amazonaws.com`
+harbourbridge data -session=mydb.session.json -source=dynamodb -source-profile="aws-access-key-id=<>,..." -target-profile="instance=my-spanner-instance,..."
 ```
 
-To use the tool on a DynamoDB database (it will migrate all tables),
-run
+You can also run HarbourBridge in a schema-and-data mode, where it will perform both
+schema and data migration. This is useful for quick evaluation when source
+database size is small.
 
 ```sh
-harbourbridge -driver=dynamodb
+harbourbridge schema-and-data -source=dynamodb -source-profile="aws-access-key-id=<>,..." -target-profile="instance=my-spanner-instance,..."
 ```
 
-Due the schemaless nature of DynamoDB, the tool infers the schema based on a
-certain amount of sampled data, by default, 100000 rows. If a table has more
-rows than the default value, we only use 100000 rows for estimating the schema.
-You can change this value via the flag `schema-sample-size`. For example,
+HarbourBridge generates a report file, a schema file, and a bad-data file (if
+there are bad-data rows). You can control where these files are written by
+specifying a file prefix. For example,
+
+```sh
+harbourbridge schema -prefix=mydb. -source=dynamodb -source-profile="aws-access-key-id=<>,..."
+```
+
+will write files `mydb.report.txt`, `mydb.schema.txt`, and
+`mydb.dropped.txt`. The prefix can also be a directory. For example,
+
+```sh
+harbourbridge schema -prefix=~/spanner-eval-mydb/ -source=dynamodb -source-profile="aws-access-key-id=<>,..."
+```
+
+would write the files into the directory `~/spanner-eval-mydb/`. Note
+that HarbourBridge will not create directories as it writes these files.
+
+HarbourBridge accepts an additional param `schema-sample-size` for
+`-source-profile` for DynamoDB. Due to the schemaless nature of DynamoDB, the
+tool infers the schema based on a certain amount of sampled data, by default,
+100000 rows. If a table has more rows than the default value, we only use
+100000 rows for estimating the schema. This flag lets you specify the number
+of rows to use for inferring schema. The default value is 100,000.
+
+Sample usage:
 
 ```sh
-harbourbridge -driver=dynamodb -schema-sample-size=500000
+harbourbridge schema -source=dynamodb -source-profile="schema-sample-size=500000,aws-access-key-id=<>,..."
 ```
 
 ## Schema Conversion
 
 The HarbourBridge tool maps DynamoDB types to Spanner types as follows:
 
-| DynamoDB Type    | Spanner Type               | Notes                                     |
+| DynamoDB Type      | Spanner Type               | Notes                                     |
 | ------------------ | -------------------------- | ----------------------------------------- |
 | `Number`           | `NUMERIC` or `STRING`      | defaults to NUMERIC, otherwise, STRING    |
 | `String`           | `STRING`                   |                                           |
@@ -98,7 +147,7 @@ and if it consistently fails, we choose STRING type for the column.
 
 In DynamoDB, a column can have a Null data type that represents an unknown or
 undefined state. Also, each row defines its own schema for columns (not for
-primary keys). So columns can be absent in rows. 
+primary keys). So columns can be absent in rows.
 
 We treat the above two cases the same as a Null value in Cloud Spanner. The
 cases that a column contains a Null value or a column is not present is an
@@ -109,7 +158,7 @@ indication that this column should be nullable.
 In Cloud Spanner, the most similar type to List and Map is
 [STRUCT](https://cloud.google.com/spanner/docs/data-types#struct_type), but it
 is not a valid column type (available for query but not for storage).
-Therefore, we encode them into a json string. 
+Therefore, we encode them into a json string.
 
 #### Occasional Errors
 
@@ -129,7 +178,7 @@ In the migration, we define a conflicting threshold on rows (after removing Null
 data types and rows that the column is not present). By default, the conflicting
 threshold is 5% and if the percentages of two or more data types are greater
 than it, we would consider that the column has conflicting data types. As a safe
-choice, we define this column as a STRING type in Cloud Spanner. 
+choice, we define this column as a STRING type in Cloud Spanner.
 
 ## Data Conversion
 
@@ -142,6 +191,6 @@ make a subsequent call to continue retrieving data from the table.
 The row result contains the data type and data itself. According to our
 [inferred schema](#schema-inference), we will parse the row to a format that
 Cloud Spanner can support. If the value parsing fails, we would drop the entire
-row and record it as bad data in the report. If a column does not appear or 
-column has a NULL data type, we would process this as a NULL value in 
-Cloud Spanner. 
+row and record it as bad data in the report. If a column does not appear or
+column has a NULL data type, we would process this as a NULL value in
+Cloud Spanner.
diff --git a/sources/dynamodb/data.go b/sources/dynamodb/data.go
index cba4728..d2740bc 100644
--- a/sources/dynamodb/data.go
+++ b/sources/dynamodb/data.go
@@ -19,74 +19,20 @@ import (
 	"fmt"
 	"math/big"
 
-	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
-// ProcessData performs data conversion for DynamoDB database. For each table,
-// we extract data using Scan requests, convert the data to Spanner data (based
-// on the source and Spanner schemas), and write it to Spanner. If we can't
-// get/process data for a table, we skip that table and process the remaining
-// tables.
-func ProcessData(conv *internal.Conv, client dynamoClient) error {
-	for srcTable, srcSchema := range conv.SrcSchema {
-		spTable, err1 := internal.GetSpannerTable(conv, srcTable)
-		spCols, err2 := internal.GetSpannerCols(conv, srcTable, srcSchema.ColNames)
-		spSchema, ok := conv.SpSchema[spTable]
-		if err1 != nil || err2 != nil || !ok {
-			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't get cols and schemas for table %s: err1=%s, err2=%s, ok=%t",
-				srcTable, err1, err2, ok))
-			continue
-		}
-
-		err := scan(srcTable, client, func(m map[string]*dynamodb.AttributeValue) {
-			spVals, badCols, srcStrVals := cvtRow(m, srcSchema, spSchema, spCols)
-			if len(badCols) == 0 {
-				conv.WriteRow(srcTable, spTable, spCols, spVals)
-			} else {
-				conv.Unexpected(fmt.Sprintf("Data conversion error for table %s in column(s) %s\n", srcTable, badCols))
-				conv.StatsAddBadRow(srcTable, conv.DataMode())
-				conv.CollectBadRow(srcTable, srcSchema.ColNames, srcStrVals)
-			}
-		})
-		if err != nil {
-			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't scan the data for table %s: %s", srcTable, err))
-		}
-	}
-	return nil
-}
-
-func scan(table string, client dynamoClient, f func(map[string]*dynamodb.AttributeValue)) error {
-	var lastEvaluatedKey map[string]*dynamodb.AttributeValue
-	for {
-		// Build the query input parameters.
-		params := &dynamodb.ScanInput{
-			TableName: aws.String(table),
-		}
-		if lastEvaluatedKey != nil {
-			params.ExclusiveStartKey = lastEvaluatedKey
-		}
-
-		// Make the DynamoDB Query API call.
-		result, err := client.Scan(params)
-		if err != nil {
-			return fmt.Errorf("failed to make Query API call for table %v: %v", table, err)
-		}
-
-		// Iterate the items returned.
-		for _, attrsMap := range result.Items {
-			f(attrsMap)
-		}
-		if result.LastEvaluatedKey == nil {
-			return nil
-		}
-		// If there are more rows, then continue.
-		lastEvaluatedKey = result.LastEvaluatedKey
+func ProcessDataRow(m map[string]*dynamodb.AttributeValue, conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) {
+	spVals, badCols, srcStrVals := cvtRow(m, srcSchema, spSchema, spCols)
+	if len(badCols) == 0 {
+		conv.WriteRow(srcTable, spTable, spCols, spVals)
+	} else {
+		conv.Unexpected(fmt.Sprintf("Data conversion error for table %s in column(s) %s\n", srcTable, badCols))
+		conv.StatsAddBadRow(srcTable, conv.DataMode())
+		conv.CollectBadRow(srcTable, srcSchema.ColNames, srcStrVals)
 	}
 }
 
diff --git a/sources/dynamodb/schema.go b/sources/dynamodb/schema.go
index 2f2ec20..868cced 100644
--- a/sources/dynamodb/schema.go
+++ b/sources/dynamodb/schema.go
@@ -23,9 +23,11 @@ import (
 	sp "cloud.google.com/go/spanner"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
 const (
@@ -45,47 +47,29 @@ const (
 	conflictThreshold = float64(0.05)
 )
 
-type dynamoClient interface {
-	ListTables(input *dynamodb.ListTablesInput) (*dynamodb.ListTablesOutput, error)
-	DescribeTable(input *dynamodb.DescribeTableInput) (*dynamodb.DescribeTableOutput, error)
-	Scan(input *dynamodb.ScanInput) (*dynamodb.ScanOutput, error)
+type InfoSchemaImpl struct {
+	DynamoClient dynamodbiface.DynamoDBAPI
+	SampleSize   int64
 }
 
-// ProcessSchema performs schema conversion for source tables in a DynamoDB
-// database. Since DynamoDB is a schemaless database, this process is imprecise.
-// We obtain schema information from two sources: from the table's metadata,
-// and from analyzing a sample of the table's rows.
-func ProcessSchema(conv *internal.Conv, client dynamoClient, tables []string, sampleSize int64) error {
-	if len(tables) == 0 {
-		var err error
-		tables, err = listTables(client)
-		if err != nil {
-			return err
-		}
-		if len(tables) == 0 {
-			return fmt.Errorf("no DynamoDB table exists under this account")
-		}
-	}
-	for _, t := range tables {
-		if err := processTable(conv, client, t, sampleSize); err != nil {
-			return err
-		}
-	}
-	common.SchemaToSpannerDDL(conv, ToDdlImpl{})
-	conv.AddPrimaryKeys()
-	return nil
+func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
+	return ToDdlImpl{}
 }
 
-func listTables(client dynamoClient) ([]string, error) {
-	var tables []string
+func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
+	return *aws.String(tableName)
+}
+
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
+	var tables []common.SchemaAndName
 	input := &dynamodb.ListTablesInput{}
 	for {
-		result, err := client.ListTables(input)
+		result, err := isi.DynamoClient.ListTables(input)
 		if err != nil {
 			return nil, err
 		}
 		for _, t := range result.TableNames {
-			tables = append(tables, *t)
+			tables = append(tables, common.SchemaAndName{Name: *t})
 		}
 
 		if result.LastEvaluatedTableName == nil {
@@ -95,40 +79,79 @@ func listTables(client dynamoClient) ([]string, error) {
 	}
 }
 
-func processTable(conv *internal.Conv, client dynamoClient, table string, sampleSize int64) error {
-	dySchema := schema.Table{Name: table}
-	err := analyzeMetadata(client, &dySchema)
+func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error) {
+	stats, count, err := scanSampleData(isi.DynamoClient, isi.SampleSize, table.Name)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
-	stats, count, err := scanSampleData(client, sampleSize, dySchema.Name)
-	if err != nil {
-		return err
-	}
-	inferDataTypes(stats, count, &dySchema)
+	return inferDataTypes(stats, count, primaryKeys)
+}
 
-	// Sort column names in increasing order, because the server may return them
-	// in a random order.
-	sort.Strings(dySchema.ColNames)
-	conv.SrcSchema[table] = dySchema
-	return nil
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
+	var lastEvaluatedKey map[string]*dynamodb.AttributeValue
+	for {
+		// Build the query input parameters.
+		params := &dynamodb.ScanInput{
+			TableName: aws.String(srcTable),
+		}
+		if lastEvaluatedKey != nil {
+			params.ExclusiveStartKey = lastEvaluatedKey
+		}
+
+		// Make the DynamoDB Query API call.
+		result, err := isi.DynamoClient.Scan(params)
+		if err != nil {
+			return nil, fmt.Errorf("failed to make Query API call for table %v: %v", srcTable, err)
+		}
+
+		if result.LastEvaluatedKey == nil {
+			return result.Items, nil
+		}
+		// If there are more rows, then continue.
+		lastEvaluatedKey = result.LastEvaluatedKey
+	}
 }
 
-func analyzeMetadata(client dynamoClient, s *schema.Table) error {
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
 	input := &dynamodb.DescribeTableInput{
-		TableName: aws.String(s.Name),
+		TableName: aws.String(table.Name),
 	}
+	result, err := isi.DynamoClient.DescribeTable(input)
+	if err != nil {
+		return 0, err
+	}
+	return *result.Table.ItemCount, err
+}
 
-	result, err := client.DescribeTable(input)
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) (primaryKeys []string, constraints map[string][]string, err error) {
+	input := &dynamodb.DescribeTableInput{
+		TableName: aws.String(table.Name),
+	}
+	result, err := isi.DynamoClient.DescribeTable(input)
 	if err != nil {
-		return fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", s.Name, err)
+		return primaryKeys, constraints, fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", table.Name, err)
 	}
 
 	// Primary keys.
 	for _, i := range result.Table.KeySchema {
-		s.PrimaryKeys = append(s.PrimaryKeys, schema.Key{Column: *i.AttributeName})
+		primaryKeys = append(primaryKeys, *i.AttributeName)
 	}
+	return primaryKeys, constraints, nil
+}
+
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+	return foreignKeys, err
+}
 
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) (indexes []schema.Index, err error) {
+	input := &dynamodb.DescribeTableInput{
+		TableName: aws.String(table.Name),
+	}
+
+	result, err := isi.DynamoClient.DescribeTable(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to make a DescribeTable API call for table %v: %v", table.Name, err)
+	}
 	// DynamoDB supports 2 types of indexes: Global Secondary Indexes (GSI) and Local Secondary Indexes (LSI).
 	// In GSI, dydb creates another global table for indexes that scales seperately.
 	// As for LSI, every partition in dydb maintains its own local index for that partition.
@@ -137,14 +160,31 @@ func analyzeMetadata(client dynamoClient, s *schema.Table) error {
 
 	// Convert secondary indexes from GlobalSecondaryIndexes.
 	for _, i := range result.Table.GlobalSecondaryIndexes {
-		s.Indexes = append(s.Indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
+		indexes = append(indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
 	}
 
 	// Convert secondary indexes from LocalSecondaryIndexes.
 	for _, i := range result.Table.LocalSecondaryIndexes {
-		s.Indexes = append(s.Indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
+		indexes = append(indexes, getSchemaIndexStruct(*i.IndexName, i.KeySchema))
 	}
+	return indexes, nil
+}
 
+// ProcessData performs data conversion for DynamoDB database. For each table,
+// we extract data using Scan requests, convert the data to Spanner data (based
+// on the source and Spanner schemas), and write it to Spanner. If we can't
+// get/process data for a table, we skip that table and process the remaining
+// tables.
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	rows, err := isi.GetRowsFromTable(conv, srcTable)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", srcTable, err))
+		return err
+	}
+	// Iterate the items returned.
+	for _, attrsMap := range rows.([]map[string]*dynamodb.AttributeValue) {
+		ProcessDataRow(attrsMap, conv, srcTable, srcSchema, spTable, spCols, spSchema)
+	}
 	return nil
 }
 
@@ -156,7 +196,7 @@ func getSchemaIndexStruct(indexName string, keySchema []*dynamodb.KeySchemaEleme
 	return schema.Index{Name: indexName, Keys: keys}
 }
 
-func scanSampleData(client dynamoClient, sampleSize int64, table string) (map[string]map[string]int64, int64, error) {
+func scanSampleData(client dynamodbiface.DynamoDBAPI, sampleSize int64, table string) (map[string]map[string]int64, int64, error) {
 	// A map from column name to a count map of possible data types.
 	stats := make(map[string]map[string]int64)
 	var count int64
@@ -248,10 +288,9 @@ type statItem struct {
 	Count int64
 }
 
-func inferDataTypes(stats map[string]map[string]int64, rows int64, s *schema.Table) {
-	if s.ColDefs == nil {
-		s.ColDefs = make(map[string]schema.Column)
-	}
+func inferDataTypes(stats map[string]map[string]int64, rows int64, primaryKeys []string) (map[string]schema.Column, []string, error) {
+	colDefs := make(map[string]schema.Column)
+	var colNames []string
 
 	for col, countMap := range stats {
 		var statItems, candidates []statItem
@@ -273,8 +312,8 @@ func inferDataTypes(stats map[string]map[string]int64, rows int64, s *schema.Tab
 
 		// Check if the column is a part of a primary key.
 		isPKey := false
-		for _, pk := range s.PrimaryKeys {
-			if pk.Column == col {
+		for _, pk := range primaryKeys {
+			if pk == col {
 				isPKey = true
 				break
 			}
@@ -294,16 +333,20 @@ func inferDataTypes(stats map[string]map[string]int64, rows int64, s *schema.Tab
 			}
 		}
 
-		s.ColNames = append(s.ColNames, col)
+		colNames = append(colNames, col)
 		if len(candidates) == 1 {
-			s.ColDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: candidates[0].Type}, NotNull: !nullable}
+			colDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: candidates[0].Type}, NotNull: !nullable}
 		} else {
 			// If there is no any candidate or more than a single candidate,
 			// this column has a significant conflict on data types and then
 			// defaults to a String type.
-			s.ColDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: typeString}, NotNull: !nullable}
+			colDefs[col] = schema.Column{Name: col, Type: schema.Type{Name: typeString}, NotNull: !nullable}
 		}
 	}
+	// Sort column names in increasing order, because the server may return them
+	// in a random order.
+	sort.Strings(colNames)
+	return colDefs, colNames, nil
 }
 
 // numericParsable determines whether its argument is a valid Spanner numeric
@@ -340,33 +383,3 @@ func numericParsable(n string) bool {
 
 	return true
 }
-
-// SetRowStats populates conv with the number of rows in each table. In
-// DynamoDB, we use describe_table api to get the number of total rows, but this
-// number is updated approximately every six hours. This means that our row
-// count could be out of date by up to six hours. One of the primary uses of
-// row count is for calculating progress during data conversion. As a result, if
-// there have been huge changes in the number of rows in a table over the last
-// six hours, the progress calculation could be inaccurate.
-func SetRowStats(conv *internal.Conv, client dynamoClient) {
-	tables, err := listTables(client)
-	if err != nil {
-		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
-		return
-	}
-	if len(tables) == 0 {
-		conv.Unexpected("no DynamoDB table exists under this account")
-		return
-	}
-	for _, t := range tables {
-		input := &dynamodb.DescribeTableInput{
-			TableName: aws.String(t),
-		}
-		result, err := client.DescribeTable(input)
-		if err != nil {
-			conv.Unexpected(fmt.Sprintf("failed to make a DescribeTable API call for table %v: %v", t, err))
-			return
-		}
-		conv.Stats.Rows[t] = *result.Table.ItemCount
-	}
-}
diff --git a/sources/dynamodb/toddl.go b/sources/dynamodb/toddl.go
index eb977e6..d1f47ec 100644
--- a/sources/dynamodb/toddl.go
+++ b/sources/dynamodb/toddl.go
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// Package dynamodb handles schema and data migrations from DynamoDB.
 package dynamodb
 
 import (
diff --git a/sources/mysql/README.md b/sources/mysql/README.md
index 4cbe9ba..13a6460 100644
--- a/sources/mysql/README.md
+++ b/sources/mysql/README.md
@@ -1,49 +1,50 @@
-# HarbourBridge: Turnkey MySQL-to-Spanner Evaluation
+# HarbourBridge: MySQL-to-Spanner Evaluation and Migration
 
-HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
-using data from an existing PostgreSQL or MySQL database. This README provides
+HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation and migration,
+using data from an existing database. This README provides
 details of the tool's MySQL capabilities. For general HarbourBridge information
-see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-turnkey-spanner-evaluation).
+see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-spanner-evaluation-and-migration).
 
 ## Example MySQL Usage
 
-The following examples assume `harbourbridge` alias has been setup as
-following.
-
-```sh
-git clone https://github.com/cloudspannerecosystem/harbourbridge
-cd harbourbridge
-alias harbourbridge="go run github.com/cloudspannerecosystem/harbourbridge"
-```
-
 HarbourBridge can either be used with mysqldump or it can be run directly
 on a MySQL database (via go's database/sql package).
 
+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.
+
 ### Using HarbourBridge with mysqldump
 
-To use HarbourBridge on a MySQL database called mydb using mysqldump, run:
+The tool can used to migrate schema from an existing mysqldump file:
 
 ```sh
-mysqldump mydb | harbourbridge -driver=mysqldump
+harbourbridge schema -source=mysql < my_mysqldump_file
 ```
 
-The tool can also be applied to an existing mysqldump file:
+This will generate a session file with `session.json` suffix. This file contains
+schema mapping from source to destination. You will need to specify this file
+during data migration. You can also specify a particular Spanner instance to use
+during data migration.
+
+For example, run
 
 ```sh
-harbourbridge -driver=mysqldump < my_mysqldump_file
+harbourbridge data -session=mydb.session.json -source=mysql -target-profile="instance=my-spanner-instance" < my_mysqldump_file
 ```
 
-To specify a particular Spanner instance to use, run:
+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
-mysqldump mydb | harbourbridge -driver=mysqldump -instance my-spanner-instance
+harbourbridge data -session=mydb.session.json -source=mysql -target-profile="instance=my-spanner-instance,dbname=my-spanner-database-name" < my_mysqldump_file
 ```
 
-By default, HarbourBridge will generate a new Spanner database name to populate.
-You can override this and specify the database name to use by:
+You can also run HarbourBridge in a schema-and-data mode, where it will perform both
+schema and data migration. This is useful for quick evaluation when source
+database size is small.
 
 ```sh
-mysqldump mydb | harbourbridge -driver=mysqldump -dbname my-spanner-database-name
+harbourbridge schema-and-data -source=mysql -target-profile="instance=my-spanner-instance" < my_mysqldump_file
 ```
 
 HarbourBridge generates a report file, a schema file, and a bad-data file (if
@@ -51,14 +52,14 @@ there are bad-data rows). You can control where these files are written by
 specifying a file prefix. For example,
 
 ```sh
-mysqldump mydb | harbourbridge -prefix mydb. -driver=mysqldump
+harbourbridge schema -prefix=mydb. -source=mysql < my_mysqldump_file
 ```
 
 will write files `mydb.report.txt`, `mydb.schema.txt`, and
 `mydb.dropped.txt`. The prefix can also be a directory. For example,
 
 ```sh
-mysqldump mydb | harbourbridge -prefix ~/spanner-eval-mydb/ -driver=mysqldump
+harbourbridge schema -prefix=~/spanner-eval-mydb/ -source=mysql < my_mysqldump_file
 ```
 
 would write the files into the directory `~/spanner-eval-mydb/`. Note
@@ -66,18 +67,25 @@ that HarbourBridge will not create directories as it writes these files.
 
 ### Directly connecting to a MySQL database
 
-To use the tool directly on a MySQL database called mydb, run
+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`.
+
+For example to perform schema conversion, run
 
 ```sh
-harbourbridge -driver=mysql
+harbourbridge schema -source=mysql -source-profile="host=<>,port=<>,user=<>,db_name=<>"
 ```
 
-It is assumed that _MYSQLHOST_, _MYSQLPORT_, _MYSQLUSER_, _MYSQLDATABASE_ environment
-variables are set. Password can be specified either in the _MYSQLPWD_ environment
-variable or provided at the password prompt.
+Parameters `port` and `password` are optional. Port (`port`) defaults to `3306`
+for MySQL source. Password can be provided at the password prompt.
+
+(⚠ Deprecated ⚠) Set environment variables `MYSQLHOST`, `MYSQLPORT`,
+`MYSQLUSER`, `MYSQLDATABASE`. Password can be specified either in the
+`MYSQLPWD` environment variable or provided at the password prompt.
 
-Note that all of the options described in the previous section on using mysqldump
-can also be used with "-driver=mysql".
+Note that the various target-profile params described in the previous section
+are also applicable in direct connect mode.
 
 ## Schema Conversion
 
@@ -258,13 +266,13 @@ type, which supports UTF-8.
 
 As noted earlier when discussing [schema conversion of
 Spatial datatype](#spatial-datatype), Spanner does not support spatial datatypes and are
-mapped to `STRING(MAX)` Spanner type. Data conversion for spatial datatypes is different
-for both `mysql` and `mysqldump` driver.
+mapped to `STRING(MAX)` Spanner type. There are differences in data conversion for spatial
+datatypes depending on whether direct connect or a mysqldump file is used.
 
-- MySQL information schema approach (`-driver=mysql`) : Data from MySQL is fetched using
+- MySQL information schema approach (direct connect) : Data from MySQL is fetched using
   'ST_AsText(g)' function which converts a value in internal geometry format to its WKT(Well-Known Text)
   representation and returns the string result. This value will be stored as `STRING` in Spanner.
-- MySQL dump approach (`-driver=mysqldump`) : Mysqldump will have the internal geometry data in
+- MySQL dump approach : Mysqldump will have the internal geometry data in
   binary format. It cannot be converted to WKT format and there is no proper method for mysqldump
   generation of spatial datatypes also. Thus, this value will just be fetched as a `TEXT` type and
   converted to Spanner type `STRING`.
diff --git a/sources/mysql/data.go b/sources/mysql/data.go
index b0978bf..590e811 100644
--- a/sources/mysql/data.go
+++ b/sources/mysql/data.go
@@ -118,7 +118,7 @@ func convScalar(conv *internal.Conv, spannerType ddl.Type, srcTypeName string, T
 		return val, nil
 	case ddl.Timestamp:
 		return convTimestamp(srcTypeName, TimezoneOffset, val)
-	case ddl.Json:
+	case ddl.JSON:
 		return val, nil
 	default:
 		return val, fmt.Errorf("data conversion not implemented for type %v", spannerType.Name)
@@ -178,7 +178,7 @@ func convInt64(val string) (int64, error) {
 // convNumeric maps a source database string value (representing a numeric)
 // into a string representing a valid Spanner numeric.
 func convNumeric(conv *internal.Conv, val string) (interface{}, error) {
-	if conv.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
 		return spanner.PGNumeric{Numeric: val, Valid: true}, nil
 	} else {
 		r := new(big.Rat)
diff --git a/sources/mysql/infoschema.go b/sources/mysql/infoschema.go
index 460c3ec..95bf33a 100644
--- a/sources/mysql/infoschema.go
+++ b/sources/mysql/infoschema.go
@@ -28,33 +28,36 @@ import (
 	_ "github.com/lib/pq"
 )
 
-// MySQL specific implementation for InfoSchema
+// InfoSchemaImpl is MySQL specific implementation for InfoSchema.
 type InfoSchemaImpl struct {
 	DbName string
+	Db     *sql.DB
 }
 
-//Functions below implement the common.InfoSchema interface
+// GetToDdl implement the common.InfoSchema interface.
 func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
 	return ToDdlImpl{}
 }
 
+// GetTableName returns table name.
 func (isi InfoSchemaImpl) GetTableName(dbName string, tableName string) string {
 	return tableName
 }
 
-func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (*sql.Rows, error) {
-	srcSchema := conv.SrcSchema[table.Name]
+// GetRowsFromTable returns a sql Rows object for a table.
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
+	srcSchema := conv.SrcSchema[srcTable]
 	srcCols := srcSchema.ColNames
 	if len(srcCols) == 0 {
-		conv.Unexpected(fmt.Sprintf("Couldn't get source columns for table %s ", table.Name))
+		conv.Unexpected(fmt.Sprintf("Couldn't get source columns for table %s ", srcTable))
 		return nil, nil
 	}
 	// MySQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but MySQL doesn't support this. So we quote it instead.
 	colNameList := buildColNameList(srcSchema, srcCols)
-	q := fmt.Sprintf("SELECT %s FROM `%s`.`%s`;", colNameList, table.Schema, table.Name)
-	rows, err := db.Query(q)
+	q := fmt.Sprintf("SELECT %s FROM `%s`.`%s`;", colNameList, conv.SrcSchema[srcTable].Schema, srcTable)
+	rows, err := isi.Db.Query(q)
 	return rows, err
 }
 
@@ -78,7 +81,16 @@ func buildColNameList(srcSchema schema.Table, srcColName []string) string {
 	return colList[:len(colList)-1]
 }
 
-func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, rows *sql.Rows) {
+// ProcessData performs data conversion for source database.
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	rowsInterface, err := isi.GetRowsFromTable(conv, srcTable)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", srcTable, err))
+		return err
+	}
+	rows := rowsInterface.(*sql.Rows)
+	defer rows.Close()
+	srcCols, _ := rows.Columns()
 	v, scanArgs := buildVals(len(srcCols))
 	for rows.Next() {
 		// get RawBytes from data.
@@ -92,15 +104,16 @@ func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string,
 		values := valsToStrings(v)
 		ProcessDataRow(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, values)
 	}
+	return nil
 }
 
 // GetRowCount with number of rows in each table.
-func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (int64, error) {
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
 	// MySQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but MySQL doesn't support this. So we quote it instead.
 	q := fmt.Sprintf("SELECT COUNT(*) FROM `%s`.`%s`;", table.Schema, table.Name)
-	rows, err := db.Query(q)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return 0, err
 	}
@@ -113,14 +126,14 @@ func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (i
 	return 0, nil //Check if 0 is ok to return
 }
 
-// getTables return list of tables in the selected database.
+// GetTables return list of tables in the selected database.
 // Note that sql.DB already effectively has the dbName
 // embedded within it (dbName is part of the DSN passed to sql.Open),
 // but unfortunately there is no way to extract it from sql.DB.
-func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error) {
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
 	// In MySQL, schema is the same as database name.
 	q := "SELECT table_name FROM information_schema.tables where table_type = 'BASE TABLE' and table_schema=?"
-	rows, err := db.Query(q, isi.DbName)
+	rows, err := isi.Db.Query(q, isi.DbName)
 	if err != nil {
 		return nil, fmt.Errorf("couldn't get tables: %w", err)
 	}
@@ -134,14 +147,15 @@ func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error)
 	return tables, nil
 }
 
-func (isi InfoSchemaImpl) GetColumns(table common.SchemaAndName, db *sql.DB) (*sql.Rows, error) {
+// GetColumns returns a list of Column objects and names// ProcessColumns
+func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error) {
 	q := `SELECT c.column_name, c.data_type, c.column_type, c.is_nullable, c.column_default, c.character_maximum_length, c.numeric_precision, c.numeric_scale, c.extra
               FROM information_schema.COLUMNS c
               where table_schema = ? and table_name = ? ORDER BY c.ordinal_position;`
-	return db.Query(q, table.Schema, table.Name)
-}
-
-func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string) {
+	cols, err := isi.Db.Query(q, table.Schema, table.Name)
+	if err != nil {
+		return nil, nil, fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
+	}
 	colDefs := make(map[string]schema.Column)
 	var colNames []string
 	var colName, dataType, isNullable, columnType string
@@ -177,20 +191,20 @@ func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, co
 		colDefs[colName] = c
 		colNames = append(colNames, colName)
 	}
-	return colDefs, colNames
+	return colDefs, colNames, nil
 }
 
-// getConstraints returns a list of primary keys and by-column map of
+// GetConstraints returns a list of primary keys and by-column map of
 // other constraints.  Note: we need to preserve ordinal order of
 // columns in primary key constraints.
 // Note that foreign key constraints are handled in getForeignKeys.
-func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]string, map[string][]string, error) {
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) ([]string, map[string][]string, error) {
 	q := `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
               FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
                 INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
                   ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA AND t.TABLE_NAME=k.TABLE_NAME
               WHERE k.TABLE_SCHEMA = ? AND k.TABLE_NAME = ? ORDER BY k.ordinal_position;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -218,12 +232,12 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table
 	return primaryKeys, m, nil
 }
 
-// getForeignKeys return list all the foreign keys constraints.
+// GetForeignKeys return list all the foreign keys constraints.
 // MySQL supports cross-database foreign key constraints. We ignore
 // them because HarbourBridge works database at a time (a specific run
 // of HarbourBridge focuses on a specific database) and so we can't handle
 // them effectively.
-func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
 	q := `SELECT k.REFERENCED_TABLE_NAME,k.COLUMN_NAME,k.REFERENCED_COLUMN_NAME,k.CONSTRAINT_NAME
 		FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t 
 		INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k 
@@ -238,7 +252,7 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
 			k.REFERENCED_TABLE_NAME,
 			k.COLUMN_NAME,
 			k.ORDINAL_POSITION;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
@@ -275,15 +289,15 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
 	return foreignKeys, nil
 }
 
-// getIndexes return a list of all indexes for the specified table.
-func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]schema.Index, error) {
+// GetIndexes return a list of all indexes for the specified table.
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
 	q := `SELECT DISTINCT INDEX_NAME,COLUMN_NAME,SEQ_IN_INDEX,COLLATION,NON_UNIQUE
 		FROM INFORMATION_SCHEMA.STATISTICS 
 		WHERE TABLE_SCHEMA = ?
 			AND TABLE_NAME = ?
 			AND INDEX_NAME != 'PRIMARY' 
 		ORDER BY INDEX_NAME, SEQ_IN_INDEX;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
diff --git a/sources/mysql/mysqldump.go b/sources/mysql/mysqldump.go
index 2645036..1faf29a 100644
--- a/sources/mysql/mysqldump.go
+++ b/sources/mysql/mysqldump.go
@@ -47,13 +47,15 @@ var spatialRegexps = func() []*regexp.Regexp {
 var spatialIndexRegex = regexp.MustCompile("(?i)\\sSPATIAL\\s")
 var spatialSridRegex = regexp.MustCompile("(?i)\\sSRID\\s\\d*")
 
+// DbDumpImpl MySQL specific implementation for DdlDumpImpl.
 type DbDumpImpl struct{}
 
-//Functions below implement the common.DbDump interface
+// GetToDdl function below implement the common.DbDump interface.
 func (ddi DbDumpImpl) GetToDdl() common.ToDdl {
 	return ToDdlImpl{}
 }
 
+// ProcessDump processes the mysql dump.
 func (ddi DbDumpImpl) ProcessDump(conv *internal.Conv, r *internal.Reader) error {
 	return processMySQLDump(conv, r)
 }
@@ -161,7 +163,7 @@ func processStatement(conv *internal.Conv, stmt ast.StmtNode) bool {
 
 func processCreateIndex(conv *internal.Conv, stmt *ast.CreateIndexStmt) {
 	if stmt.Table == nil {
-		logStmtError(conv, stmt, fmt.Errorf("cannot process index statement with nil table."))
+		logStmtError(conv, stmt, fmt.Errorf("cannot process index statement with nil table"))
 		return
 	}
 	tableName, err := getTableName(stmt.Table)
diff --git a/sources/mysql/toddl.go b/sources/mysql/toddl.go
index 72505cd..e76a5cf 100644
--- a/sources/mysql/toddl.go
+++ b/sources/mysql/toddl.go
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// Package MySQL handles schema and data migrations from MySQL.
 package mysql
 
 import (
@@ -21,18 +22,18 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
-// MySQL specific implementation for ToDdl
+// ToDdlImpl MySQL specific implementation for ToDdl.
 type ToDdlImpl struct {
 }
 
-// Functions below implement the common.ToDdl interface
-// toSpannerType maps a scalar source schema type (defined by id and
+// ToSpannerType maps a scalar source schema type (defined by id and
 // mods) into a Spanner type. This is the core source-to-Spanner type
 // mapping.  toSpannerType returns the Spanner type and a list of type
 // conversion issues encountered.
+// Functions below implement the common.ToDdl interface
 func (tdi ToDdlImpl) ToSpannerType(conv *internal.Conv, columnType schema.Type) (ddl.Type, []internal.SchemaIssue) {
 	ty, issues := toSpannerTypeInternal(conv, columnType.Name, columnType.Mods)
-	if conv.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
 		ty = overrideExperimentalType(columnType, ty)
 	} else {
 		if len(columnType.ArrayBounds) > 1 {
@@ -83,7 +84,7 @@ func toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Ty
 	case "set", "enum":
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
 	case "json":
-		return ddl.Type{Name: ddl.Json}, nil
+		return ddl.Type{Name: ddl.JSON}, nil
 	case "binary", "varbinary":
 		return ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, nil
 	case "tinyblob", "mediumblob", "blob", "longblob":
diff --git a/sources/postgres/README.md b/sources/postgres/README.md
index d2cbd8d..a6c35c6 100644
--- a/sources/postgres/README.md
+++ b/sources/postgres/README.md
@@ -1,49 +1,53 @@
-# HarbourBridge: Turnkey PostgreSQL-to-Spanner Evaluation
+# HarbourBridge: PostgreSQL-to-Spanner Evaluation and Migration
 
 HarbourBridge is a stand-alone open source tool for Cloud Spanner evaluation,
 using data from an existing PostgreSQL or MySQL database. This README provides
 details of the tool's PostgreSQL capabilities. For general HarbourBridge information
-see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-turnkey-spanner-evaluation).
+see this [README](https://github.com/cloudspannerecosystem/harbourbridge#harbourbridge-spanner-evaluation-and-migration).
 
 ## Example PostgreSQL Usage
 
-The following examples assume `harbourbridge` alias has been setup as
-following.
-
-```sh
-git clone https://github.com/cloudspannerecosystem/harbourbridge
-cd harbourbridge
-alias harbourbridge="go run github.com/cloudspannerecosystem/harbourbridge"
-```
-
 HarbourBridge can either be used with pg_dump or it can be run directly
 on a PostgreSQL database (via go's database/sql package).
 
+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.
+
 ### Using HarbourBridge with pg_dump
 
-To use HarbourBridge on a PostgreSQL database called mydb using pg_dump output,run:
+The tool can used to migrate schema from an existing pg_dump file:
 
 ```sh
-pg_dump mydb | harbourbridge -driver=pg_dump
+harbourbridge schema -source=postgresql < my_pg_dump_file
 ```
 
-The tool can also be applied to an existing pg_dump file:
+You can use any of `postgresql`, `postgres`, or `pg` as the argument to the
+`-source` flag. They all specify PostgreSQL as the source database.
 
-```sh
-harbourbridge -driver=pg_dump < my_pg_dump_file
-```
+This will generate a session file with `session.json` suffix. This file contains
+schema mapping from source to destination. You will need to specify this file
+during data migration. You can also specify a particular Spanner instance to use
+during data migration.
 
-To specify a particular Spanner instance to use, run:
+For example, run
 
 ```sh
-pg_dump mydb | harbourbridge -driver=pg_dump -instance my-spanner-instance
+harbourbridge data -session=mydb.session.json -source=pg -target-profile="instance=my-spanner-instance" < my_pg_dump_file
 ```
 
 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
-pg_dump mydb | harbourbridge -driver=pg_dump -dbname my-spanner-database-name
+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
+schema and data migration. This is useful for quick evaluation when source
+database size is small.
+
+```sh
+harbourbridge schema-and-data -source=pg -target-profile="instance=my-spanner-instance" < my_pg_dump_file
 ```
 
 HarbourBridge generates a report file, a schema file, and a bad-data file (if
@@ -51,14 +55,14 @@ there are bad-data rows). You can control where these files are written by
 specifying a file prefix. For example,
 
 ```sh
-pg_dump mydb | harbourbridge -driver=pg_dump -prefix mydb.
+harbourbridge schema -prefix=mydb. -source=postgres < my_pg_dump_file
 ```
 
 will write files `mydb.report.txt`, `mydb.schema.txt`, and
 `mydb.dropped.txt`. The prefix can also be a directory. For example,
 
 ```sh
-pg_dump mydb | harbourbridge -driver=pg_dump -prefix ~/spanner-eval-mydb/
+harbourbridge schema -prefix=~/spanner-eval-mydb/ -source=postgres < my_pg_dump_file
 ```
 
 would write the files into the directory `~/spanner-eval-mydb/`. Note
@@ -66,18 +70,26 @@ that HarbourBridge will not create directories as it writes these files.
 
 ### Directly connecting to a PostgreSQL database
 
-To use the tool directly on a PostgresSQL database called mydb, run
+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
+`password`.
+
+For example to perform schema conversion, run
 
 ```sh
-harbourbridge -driver=postgres
+harbourbridge schema -source=postgres -source-profile="host=<>,port=<>,user=<>,db_name=<>"
 ```
 
-It is assumed that _PGHOST_, _PGPORT_, _PGUSER_, _PGDATABASE_ environment
-variables are set. Password can be specified either in the _PGPASSWORD_ environment
-variable or provided at the password prompt.
+Parameters `port` and `password` are optional. Port (`port`) defaults to `5432`
+for PostgreSQL source. Password can be provided at the password prompt.
+
+Alternatively, you can also set environment variables `PGHOST`, `PGPORT`,
+`PGUSER`, `PGDATABASE` for direct access. Password can be specified either in
+the `PGPASSWORD` environment variable or provided at the password prompt.
 
-Note that all of the options described in the previous section on using pg_dump can
-also be used with "-driver=postgres".
+Note that the various target-profile params described in the previous section
+are also applicable in direct connect mode.
 
 ## Schema Conversion
 
diff --git a/sources/postgres/data.go b/sources/postgres/data.go
index d799229..da3d237 100644
--- a/sources/postgres/data.go
+++ b/sources/postgres/data.go
@@ -133,7 +133,7 @@ func convScalar(conv *internal.Conv, spannerType ddl.Type, srcTypeName string, l
 		return val, nil
 	case ddl.Timestamp:
 		return convTimestamp(srcTypeName, location, val)
-	case ddl.Json:
+	case ddl.JSON:
 		return val, nil
 	default:
 		return val, fmt.Errorf("data conversion not implemented for type %v", spannerType.Name)
@@ -186,7 +186,7 @@ func convInt64(val string) (int64, error) {
 // convNumeric maps a source database string value (representing a numeric)
 // into a string representing a valid Spanner numeric.
 func convNumeric(conv *internal.Conv, val string) (interface{}, error) {
-	if conv.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
 		return spanner.PGNumeric{Numeric: val, Valid: true}, nil
 	} else {
 		r := new(big.Rat)
diff --git a/sources/postgres/infoschema.go b/sources/postgres/infoschema.go
index 79f345c..cfea57a 100644
--- a/sources/postgres/infoschema.go
+++ b/sources/postgres/infoschema.go
@@ -32,15 +32,17 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
-// Postgres specific implementation for InfoSchema
+// InfoSchemaImpl postgres specific implementation for InfoSchema.
 type InfoSchemaImpl struct {
+	Db *sql.DB
 }
 
-//Functions below implement the common.InfoSchema interface
+// GetToDdl function below implement the common.InfoSchema interface.
 func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
 	return ToDdlImpl{}
 }
 
+// GetTableName returns table name.
 func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
 	if schema == "public" { // Drop 'public' prefix.
 		return tableName
@@ -48,19 +50,20 @@ func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
 	return fmt.Sprintf("%s.%s", schema, tableName)
 }
 
-func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (*sql.Rows, error) {
+// GetRowsFromTable returns a sql Rows object for a table.
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
 	// PostgreSQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but PostgreSQL doesn't support this. So we quote it instead.
-	q := fmt.Sprintf(`SELECT * FROM "%s"."%s";`, table.Schema, table.Name)
-	rows, err := db.Query(q)
+	q := fmt.Sprintf(`SELECT * FROM "%s"."%s";`, conv.SrcSchema[srcTable].Schema, srcTable)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return nil, err
 	}
 	return rows, err
 }
 
-// ProcessSQLData performs data conversion for source database
+// ProcessDataRows performs data conversion for source database
 // 'db'. For each table, we extract data using a "SELECT *" query,
 // convert the data to Spanner data (based on the source and Spanner
 // schemas), and write it to Spanner.  If we can't get/process data
@@ -79,7 +82,15 @@ func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, db *sql.DB, tabl
 // We choose to do all type conversions explicitly ourselves so that
 // we can generate more targeted error messages: hence we pass
 // *interface{} parameters to row.Scan.
-func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, rows *sql.Rows) {
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	rowsInterface, err := isi.GetRowsFromTable(conv, srcTable)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", srcTable, err))
+		return err
+	}
+	rows := rowsInterface.(*sql.Rows)
+	defer rows.Close()
+	srcCols, _ := rows.Columns()
 	v, iv := buildVals(len(srcCols))
 	for rows.Next() {
 		err := rows.Scan(iv...)
@@ -98,6 +109,7 @@ func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string,
 		}
 		conv.WriteRow(srcTable, spTable, cvtCols, cvtVals)
 	}
+	return nil
 }
 
 // ConvertSQLRow performs data conversion for a single row of data
@@ -141,12 +153,12 @@ func convertSQLRow(conv *internal.Conv, srcTable string, srcCols []string, srcSc
 }
 
 // GetRowCount with number of rows in each table.
-func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (int64, error) {
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
 	// PostgreSQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but PostgreSQL doesn't support this. So we quote it instead.
 	q := fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s";`, table.Schema, table.Name)
-	rows, err := db.Query(q)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return 0, err
 	}
@@ -159,18 +171,19 @@ func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (i
 	return 0, nil //Check if 0 is ok to return
 }
 
+// GetTables return list of tables in the selected database.
 // TODO: All of the queries to get tables and table data should be in
 // a single transaction to ensure we obtain a consistent snapshot of
 // schema information and table data (pg_dump does something
 // similar).
-func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error) {
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
 	ignored := make(map[string]bool)
 	// Ignore all system tables: we just want to convert user tables.
 	for _, s := range []string{"information_schema", "postgres", "pg_catalog", "pg_temp_1", "pg_toast", "pg_toast_temp_1"} {
 		ignored[s] = true
 	}
 	q := "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'"
-	rows, err := db.Query(q)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return nil, fmt.Errorf("couldn't get tables: %w", err)
 	}
@@ -186,16 +199,17 @@ func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error)
 	return tables, nil
 }
 
-func (isi InfoSchemaImpl) GetColumns(table common.SchemaAndName, db *sql.DB) (*sql.Rows, error) {
+// GetColumns returns a list of Column objects and names
+func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error) {
 	q := `SELECT c.column_name, c.data_type, e.data_type, c.is_nullable, c.column_default, c.character_maximum_length, c.numeric_precision, c.numeric_scale
               FROM information_schema.COLUMNS c LEFT JOIN information_schema.element_types e
                  ON ((c.table_catalog, c.table_schema, c.table_name, 'TABLE', c.dtd_identifier)
                      = (e.object_catalog, e.object_schema, e.object_name, e.object_type, e.collection_type_identifier))
               where table_schema = $1 and table_name = $2 ORDER BY c.ordinal_position;`
-	return db.Query(q, table.Schema, table.Name)
-}
-
-func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string) {
+	cols, err := isi.Db.Query(q, table.Schema, table.Name)
+	if err != nil {
+		return nil, nil, fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
+	}
 	colDefs := make(map[string]schema.Column)
 	var colNames []string
 	var colName, dataType, isNullable string
@@ -229,20 +243,20 @@ func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, co
 		colDefs[colName] = c
 		colNames = append(colNames, colName)
 	}
-	return colDefs, colNames
+	return colDefs, colNames, nil
 }
 
-// getConstraints returns a list of primary keys and by-column map of
+// GetConstraints returns a list of primary keys and by-column map of
 // other constraints.  Note: we need to preserve ordinal order of
 // columns in primary key constraints.
 // Note that foreign key constraints are handled in getForeignKeys.
-func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]string, map[string][]string, error) {
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) ([]string, map[string][]string, error) {
 	q := `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
               FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
                 INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
                   ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
               WHERE k.TABLE_SCHEMA = $1 AND k.TABLE_NAME = $2 ORDER BY k.ordinal_position;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -270,8 +284,8 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table
 	return primaryKeys, m, nil
 }
 
-// getForeignKeys returns a list of all the foreign key constraints.
-func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+// GetForeignKeys returns a list of all the foreign key constraints.
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
 	q := `SELECT 
 		schema_name AS "TABLE_SCHEMA", 
 		cl.relname AS "TABLE_NAME", 
@@ -296,7 +310,7 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
    		JOIN PG_ATTRIBUTE att2 ON
        		att2.attrelid = con.conrelid AND att2.attnum = con.parent;`
 
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
@@ -335,11 +349,11 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
 	return foreignKeys, nil
 }
 
-// getIndexes return a list of all indexes for the specified table.
+// GetIndexes return a list of all indexes for the specified table.
 // Note: Extracting index definitions from PostgreSQL information schema tables is complex.
 // See https://stackoverflow.com/questions/6777456/list-all-index-names-column-names-and-its-table-name-of-a-postgresql-database/44460269#44460269
 // for background.
-func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]schema.Index, error) {
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
 	q := `SELECT
 			irel.relname AS index_name,
 			a.attname AS column_name,
@@ -369,7 +383,7 @@ func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, db *sql.DB, table comm
            		array_position(i.indkey, a.attnum),
            		o.OPTION,i.indisunique
 		ORDER BY irel.relname, array_position(i.indkey, a.attnum);`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
@@ -512,7 +526,7 @@ func cvtSQLScalar(conv *internal.Conv, srcCd schema.Column, spCd ddl.ColumnDef,
 		case time.Time:
 			return v, nil
 		}
-	case ddl.Json:
+	case ddl.JSON:
 		switch v := val.(type) {
 		case string:
 			return string(v), nil
diff --git a/sources/postgres/pgdump.go b/sources/postgres/pgdump.go
index 944538f..4318ae8 100644
--- a/sources/postgres/pgdump.go
+++ b/sources/postgres/pgdump.go
@@ -28,6 +28,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 )
 
+// DbDumpImpl Postgres specific implementation for DdlDumpImpl.
 type DbDumpImpl struct{}
 
 type copyOrInsert struct {
@@ -44,16 +45,17 @@ const (
 	insert
 )
 
-//Functions below implement the common.DbDump interface
+// GetToDdl functions below implement the common.DbDump interface
 func (ddi DbDumpImpl) GetToDdl() common.ToDdl {
 	return ToDdlImpl{}
 }
 
+// ProcessDump calls processPgDump to read a Postgres dump file
 func (ddi DbDumpImpl) ProcessDump(conv *internal.Conv, r *internal.Reader) error {
 	return processPgDump(conv, r)
 }
 
-// ProcessPgDump reads pg_dump data from r and does schema or data conversion,
+// processPgDump reads pg_dump data from r and does schema or data conversion,
 // depending on whether conv is configured for schema mode or data mode.
 // In schema mode, ProcessPgDump incrementally builds a schema (updating conv).
 // In data mode, ProcessPgDump uses this schema to convert PostgreSQL data
diff --git a/sources/postgres/toddl.go b/sources/postgres/toddl.go
index b5c8dad..fe88787 100644
--- a/sources/postgres/toddl.go
+++ b/sources/postgres/toddl.go
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// Package postgres handles schema and data migrations from Postgres.
 package postgres
 
 import (
@@ -21,17 +22,17 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
-// Postgres specific implementation for ToDdl
+// ToDdlImpl Postgres specific implementation for ToDdl.
 type ToDdlImpl struct {
 }
 
-// toSpannerType maps a scalar source schema type (defined by id and
+// ToSpannerType maps a scalar source schema type (defined by id and
 // mods) into a Spanner type. This is the core source-to-Spanner type
 // mapping.  toSpannerType returns the Spanner type and a list of type
 // conversion issues encountered.
 func (tdi ToDdlImpl) ToSpannerType(conv *internal.Conv, columnType schema.Type) (ddl.Type, []internal.SchemaIssue) {
 	ty, issues := toSpannerTypeInternal(conv, columnType.Name, columnType.Mods)
-	if conv.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if conv.TargetDb == constants.TargetExperimentalPostgres {
 		ty = overrideExperimentalType(columnType, ty)
 	} else {
 		if len(columnType.ArrayBounds) > 1 {
@@ -102,7 +103,7 @@ func toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Ty
 		}
 		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
 	case "json", "jsonb":
-		return ddl.Type{Name: ddl.Json}, nil
+		return ddl.Type{Name: ddl.JSON}, nil
 	}
 	return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.NoGoodType}
 }
diff --git a/spanner/ddl/ast.go b/spanner/ddl/ast.go
index 33fc8b2..48ad695 100644
--- a/spanner/ddl/ast.go
+++ b/spanner/ddl/ast.go
@@ -49,7 +49,7 @@ const (
 	// Numeric represent NUMERIC type.
 	Numeric string = "NUMERIC"
 	// Json represent JSON type.
-	Json string = "JSON"
+	JSON string = "JSON"
 	// MaxLength is a sentinel for Type's Len field, representing the MAX value.
 	MaxLength = math.MaxInt64
 
@@ -168,7 +168,7 @@ func (c Config) quote(s string) string {
 // needs of PrintCreateTable.
 func (cd ColumnDef) PrintColumnDef(c Config) (string, string) {
 	var s string
-	if c.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if c.TargetDb == constants.TargetExperimentalPostgres {
 		s = fmt.Sprintf("%s %s", c.quote(cd.Name), cd.T.PGPrintColumnDefType())
 	} else {
 		s = fmt.Sprintf("%s %s", c.quote(cd.Name), cd.T.PrintColumnDefType())
@@ -267,7 +267,7 @@ func (ct CreateTable) PrintCreateTable(config Config) string {
 
 	var interleave string
 	if ct.Parent != "" {
-		if config.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+		if config.TargetDb == constants.TargetExperimentalPostgres {
 			// PG spanner only supports PRIMARY KEY() inside the CREATE TABLE()
 			// and thus INTERLEAVE follows immediately after closing brace.
 			interleave = " INTERLEAVE IN PARENT " + config.quote(ct.Parent)
@@ -276,7 +276,7 @@ func (ct CreateTable) PrintCreateTable(config Config) string {
 		}
 	}
 
-	if config.TargetDb == constants.TARGET_EXPERIMENTAL_POSTGRES {
+	if config.TargetDb == constants.TargetExperimentalPostgres {
 		return fmt.Sprintf("%sCREATE TABLE %s (\n%s\tPRIMARY KEY (%s)\n)%s", tableComment, config.quote(ct.Name), cols, strings.Join(keys, ", "), interleave)
 	}
 	return fmt.Sprintf("%sCREATE TABLE %s (\n%s) PRIMARY KEY (%s)%s", tableComment, config.quote(ct.Name), cols, strings.Join(keys, ", "), interleave)
@@ -320,8 +320,10 @@ func (k Foreignkey) PrintForeignKeyAlterTable(c Config, tableName string) string
 	return fmt.Sprintf("ALTER TABLE %s ADD %sFOREIGN KEY (%s) REFERENCES %s (%s)", c.quote(tableName), s, strings.Join(cols, ", "), c.quote(k.ReferTable), strings.Join(referCols, ", "))
 }
 
+// Schema stores a map of table names and Tables.
 type Schema map[string]CreateTable
 
+// NewSchema creates a new Schema object.
 func NewSchema() Schema {
 	return make(map[string]CreateTable)
 }
diff --git a/web/web.go b/web/web.go
index 13324c8..8a2a882 100644
--- a/web/web.go
+++ b/web/web.go
@@ -43,7 +43,6 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/gorilla/handlers"
-	_ "github.com/lib/pq"
 )
 
 // TODO:(searce):
@@ -126,9 +125,9 @@ func convertSchemaSQL(w http.ResponseWriter, r *http.Request) {
 	var err error
 	switch sessionState.driver {
 	case constants.MYSQL:
-		err = common.ProcessInfoSchema(conv, sessionState.sourceDB, mysql.InfoSchemaImpl{sessionState.dbName})
+		err = common.ProcessSchema(conv, mysql.InfoSchemaImpl{DbName: sessionState.dbName, Db: sessionState.sourceDB})
 	case constants.POSTGRES:
-		err = common.ProcessInfoSchema(conv, sessionState.sourceDB, postgres.InfoSchemaImpl{})
+		err = common.ProcessSchema(conv, postgres.InfoSchemaImpl{Db: sessionState.sourceDB})
 	default:
 		http.Error(w, fmt.Sprintf("Driver : '%s' is not supported", sessionState.driver), http.StatusBadRequest)
 		return
@@ -168,7 +167,7 @@ func convertSchemaDump(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, fmt.Sprintf("failed to open dump file %v : %v", dc.FilePath, err), http.StatusNotFound)
 		return
 	}
-	conv, err := conversion.SchemaConv(dc.Driver, "", constants.TARGET_SPANNER, &conversion.IOStreams{In: f, Out: os.Stdout}, 0)
+	conv, err := conversion.SchemaConv(dc.Driver, "", constants.TargetSpanner, &conversion.IOStreams{In: f, Out: os.Stdout}, 0)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Schema Conversion Error : %v", err), http.StatusNotFound)
 		return
@@ -342,7 +341,7 @@ func updateTableSchema(w http.ResponseWriter, r *http.Request) {
 	srcTableName := sessionState.conv.ToSource[table].Name
 	for colName, v := range t.UpdateCols {
 		if v.Removed {
-			err, status := canRemoveColumn(colName, table)
+			status, err := canRemoveColumn(colName, table)
 			if err != nil {
 				err = rollback(err)
 				http.Error(w, fmt.Sprintf("%v", err), status)
@@ -352,7 +351,7 @@ func updateTableSchema(w http.ResponseWriter, r *http.Request) {
 			continue
 		}
 		if v.Rename != "" && v.Rename != colName {
-			if err, status := canRenameOrChangeType(colName, table); err != nil {
+			if status, err := canRenameOrChangeType(colName, table); err != nil {
 				err = rollback(err)
 				http.Error(w, fmt.Sprintf("%v", err), status)
 				return
@@ -373,7 +372,7 @@ func updateTableSchema(w http.ResponseWriter, r *http.Request) {
 			}
 
 			if typeChange {
-				if err, status := canRenameOrChangeType(colName, table); err != nil {
+				if status, err := canRenameOrChangeType(colName, table); err != nil {
 					err = rollback(err)
 					http.Error(w, fmt.Sprintf("%v", err), status)
 					return
@@ -439,6 +438,7 @@ func getReportFile(w http.ResponseWriter, r *http.Request) {
 	w.Write([]byte(reportAbsPath))
 }
 
+// TableInterleaveStatus stores data regarding interleave status.
 type TableInterleaveStatus struct {
 	Possible bool
 	Parent   string
@@ -856,44 +856,43 @@ func isReferencedByFK(col, table string) (bool, string) {
 	return false, ""
 }
 
-func canRemoveColumn(colName, table string) (error, int) {
+func canRemoveColumn(colName, table string) (int, error) {
 	if isPartOfPK := isPartOfPK(colName, table); isPartOfPK {
-		return fmt.Errorf("column is part of primary key"), http.StatusBadRequest
+		return http.StatusBadRequest, fmt.Errorf("column is part of primary key")
 	}
 	if isPartOfSecondaryIndex, _ := isPartOfSecondaryIndex(colName, table); isPartOfSecondaryIndex {
-		return fmt.Errorf("column is part of secondary index, remove secondary index before making the update"), http.StatusPreconditionFailed
+		return http.StatusPreconditionFailed, fmt.Errorf("column is part of secondary index, remove secondary index before making the update")
 	}
 	isPartOfFK := isPartOfFK(colName, table)
 	isReferencedByFK, _ := isReferencedByFK(colName, table)
 	if isPartOfFK || isReferencedByFK {
-		return fmt.Errorf("column is part of foreign key relation, remove foreign key constraint before making the update"), http.StatusPreconditionFailed
+		return http.StatusPreconditionFailed, fmt.Errorf("column is part of foreign key relation, remove foreign key constraint before making the update")
 	}
-	return nil, http.StatusOK
+	return http.StatusOK, nil
 }
 
-func canRenameOrChangeType(colName, table string) (error, int) {
+func canRenameOrChangeType(colName, table string) (int, error) {
 	isPartOfPK := isPartOfPK(colName, table)
 	isParent, childSchema := isParent(table)
 	isChild := sessionState.conv.SpSchema[table].Parent != ""
 	if isPartOfPK && (isParent || isChild) {
-		return fmt.Errorf("Column : '%s' in table : '%s' is part of parent-child relation with schema : '%s'", colName, table, childSchema), http.StatusBadRequest
+		return http.StatusBadRequest, fmt.Errorf("column : '%s' in table : '%s' is part of parent-child relation with schema : '%s'", colName, table, childSchema)
 	}
 	if isPartOfSecondaryIndex, indexName := isPartOfSecondaryIndex(colName, table); isPartOfSecondaryIndex {
-		return fmt.Errorf("Column : '%s' in table : '%s' is part of secondary index : '%s', remove secondary index before making the update",
-			colName, table, indexName), http.StatusPreconditionFailed
+		return http.StatusPreconditionFailed, fmt.Errorf("column : '%s' in table : '%s' is part of secondary index : '%s', remove secondary index before making the update",
+			colName, table, indexName)
 	}
 	isPartOfFK := isPartOfFK(colName, table)
 	isReferencedByFK, relationTable := isReferencedByFK(colName, table)
 	if isPartOfFK || isReferencedByFK {
 		if isReferencedByFK {
-			return fmt.Errorf("Column : '%s' in table : '%s' is part of foreign key relation with table : '%s', remove foreign key constraint before making the update",
-				colName, table, relationTable), http.StatusPreconditionFailed
-		} else {
-			return fmt.Errorf("Column : '%s' in table : '%s' is part of foreign keys, remove foreign key constraint before making the update",
-				colName, table), http.StatusPreconditionFailed
+			return http.StatusPreconditionFailed, fmt.Errorf("column : '%s' in table : '%s' is part of foreign key relation with table : '%s', remove foreign key constraint before making the update",
+				colName, table, relationTable)
 		}
+		return http.StatusPreconditionFailed, fmt.Errorf("column : '%s' in table : '%s' is part of foreign keys, remove foreign key constraint before making the update",
+			colName, table)
 	}
-	return nil, http.StatusOK
+	return http.StatusOK, nil
 }
 
 func checkPrimaryKeyPrefix(table string, refTable string, fk ddl.Foreignkey, tableInterleaveStatus *TableInterleaveStatus) bool {
@@ -1102,6 +1101,7 @@ func getFilePrefix(now time.Time) (string, error) {
 	return dbName + ".", nil
 }
 
+// SessionState stores information for the current migration session.
 type SessionState struct {
 	sourceDB    *sql.DB        // Connection to source database in case of direct connection
 	dbName      string         // Name of source database
@@ -1160,7 +1160,8 @@ func init() {
 	sessionState.conv = internal.MakeConv()
 }
 
-func WebApp() {
+// App connects to the web app.
+func App() {
 	addr := ":8080"
 	router := getRoutes()
 	log.Printf("Starting server at port 8080\n")


Test patch:
diff --git a/cmd/source_profile_test.go b/cmd/source_profile_test.go
index b7a3a76..6562ee9 100644
--- a/cmd/source_profile_test.go
+++ b/cmd/source_profile_test.go
@@ -120,12 +120,12 @@ func TestNewSourceProfileConnectionDynamoDB(t *testing.T) {
 		},
 		{
 			name:          "valid schema sample size",
-			params:        map[string]string{"schemaSampleSize": "15"},
+			params:        map[string]string{"schema-sample-size": "15"},
 			errorExpected: false,
 		},
 		{
 			name:          "invalid schema sample size",
-			params:        map[string]string{"schemaSampleSize": "a"},
+			params:        map[string]string{"schema-sample-size": "a"},
 			errorExpected: true,
 		},
 	}
diff --git a/internal/mapping_test.go b/internal/mapping_test.go
index c739261..07424ce 100644
--- a/internal/mapping_test.go
+++ b/internal/mapping_test.go
@@ -111,7 +111,7 @@ func TestToSpannerForeignKey(t *testing.T) {
 	}
 }
 
-func TestGetSpannerId(t *testing.T) {
+func TestGetSpannerID(t *testing.T) {
 	conv := MakeConv()
 	basicTests := []struct {
 		name       string // Name of test.
@@ -134,7 +134,7 @@ func TestGetSpannerId(t *testing.T) {
 		{"Bad name with different case collision", "IN\tDex", "IN_Dex_13"},
 	}
 	for _, tc := range basicTests {
-		spKeyName := getSpannerId(conv, tc.srcKeyName)
+		spKeyName := getSpannerID(conv, tc.srcKeyName)
 		assert.Equal(t, tc.spKeyName, spKeyName, tc.name)
 	}
 }
diff --git a/sources/dynamodb/data_test.go b/sources/dynamodb/data_test.go
index ea5f7f1..7347a82 100644
--- a/sources/dynamodb/data_test.go
+++ b/sources/dynamodb/data_test.go
@@ -33,58 +33,37 @@ type spannerData struct {
 	vals  []interface{}
 }
 
-func TestProcessData(t *testing.T) {
+func TestProcessDataRow(t *testing.T) {
 	strA := "str-1"
-	strB := "str-2"
 	numStr1 := "10.1"
 	numStr2 := "12.34"
-	numStr3 := "89.0"
 	numVal1 := big.NewRat(101, 10)
-	numVal2 := big.NewRat(89, 1)
 
 	boolVal := true
-	scanOutputs := []dynamodb.ScanOutput{
+	items := []map[string]*dynamodb.AttributeValue{
 		{
-			Items: []map[string]*dynamodb.AttributeValue{
-				{
-					"a": {S: &strA},
-					"b": {N: &numStr1},
-					"c": {N: &numStr2},
-					"d": {BOOL: &boolVal},
-				},
-			},
-			LastEvaluatedKey: map[string]*dynamodb.AttributeValue{
-				"a": {S: &strA},
-			},
-		},
-		{
-			Items: []map[string]*dynamodb.AttributeValue{
-				{
-					"a": {S: &strB},
-					"b": {N: &numStr3},
-				},
-			},
+			"a": {S: &strA},
+			"b": {N: &numStr1},
+			"c": {N: &numStr2},
+			"d": {BOOL: &boolVal},
 		},
 	}
 
-	client := &mockDynamoClient{
-		scanOutputs: scanOutputs,
-	}
-
 	tableName := "testtable"
 	cols := []string{"a", "b", "c", "d"}
-	conv := buildConv(
-		ddl.CreateTable{
-			Name:     tableName,
-			ColNames: cols,
-			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}},
-			},
-			Pks: []ddl.IndexKey{{Col: "a"}},
+	spSchema := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: cols,
+		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}},
 		},
+		Pks: []ddl.IndexKey{{Col: "a"}},
+	}
+	conv := buildConv(
+		spSchema,
 		schema.Table{
 			Name:     tableName,
 			ColNames: cols,
@@ -103,7 +82,9 @@ func TestProcessData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	ProcessData(conv, client)
+	for _, attrsMap := range items {
+		ProcessDataRow(attrsMap, conv, tableName, conv.SrcSchema[tableName], tableName, cols, spSchema)
+	}
 	assert.Equal(t,
 		[]spannerData{
 			{
@@ -111,11 +92,6 @@ func TestProcessData(t *testing.T) {
 				cols:  cols,
 				vals:  []interface{}{"str-1", *numVal1, "12.34", true},
 			},
-			{
-				table: tableName,
-				cols:  cols,
-				vals:  []interface{}{"str-2", *numVal2, nil, nil},
-			},
 		},
 		rows,
 	)
diff --git a/sources/dynamodb/schema_test.go b/sources/dynamodb/schema_test.go
index 8400392..4f7a00f 100644
--- a/sources/dynamodb/schema_test.go
+++ b/sources/dynamodb/schema_test.go
@@ -16,11 +16,14 @@ package dynamodb
 
 import (
 	"fmt"
+	"math/big"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
 )
@@ -32,6 +35,7 @@ type mockDynamoClient struct {
 	describeTableOutputs   []dynamodb.DescribeTableOutput
 	scanCallCount          int
 	scanOutputs            []dynamodb.ScanOutput
+	dynamodbiface.DynamoDBAPI
 }
 
 func (m *mockDynamoClient) ListTables(input *dynamodb.ListTablesInput) (*dynamodb.ListTablesOutput, error) {
@@ -85,6 +89,23 @@ func TestProcessSchema(t *testing.T) {
 				},
 			},
 		},
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameA,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+				},
+			},
+		},
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameB,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+					{AttributeName: &attrNameB, KeyType: &sortKeyType},
+				},
+			},
+		},
 		{
 			Table: &dynamodb.TableDescription{
 				TableName: &tableNameB,
@@ -143,11 +164,10 @@ func TestProcessSchema(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 		scanOutputs:          scanOutputs,
 	}
-	tables := []string{}
 	sampleSize := int64(10000)
 
 	conv := internal.MakeConv()
-	err := ProcessSchema(conv, client, tables, sampleSize)
+	err := common.ProcessSchema(conv, InfoSchemaImpl{client, sampleSize})
 
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
@@ -209,6 +229,15 @@ func TestProcessSchema_FullDataTypes(t *testing.T) {
 				},
 			},
 		},
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameA,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+					{AttributeName: &attrNameB, KeyType: &sortKeyType},
+				},
+			},
+		},
 	}
 	scanOutputs := []dynamodb.ScanOutput{
 		{
@@ -238,11 +267,10 @@ func TestProcessSchema_FullDataTypes(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 		scanOutputs:          scanOutputs,
 	}
-	tables := []string{}
 	sampleSize := int64(10000)
 
 	conv := internal.MakeConv()
-	err := ProcessSchema(conv, client, tables, sampleSize)
+	err := common.ProcessSchema(conv, InfoSchemaImpl{client, sampleSize})
 
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
@@ -269,8 +297,76 @@ func TestProcessSchema_FullDataTypes(t *testing.T) {
 	assert.Equal(t, int64(0), conv.Unexpecteds())
 }
 
+func TestProcessData(t *testing.T) {
+	strA := "str-1"
+	numStr1 := "10.1"
+	numStr2 := "12.34"
+	numVal1 := big.NewRat(101, 10)
+
+	boolVal := true
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+					"b": {N: &numStr1},
+					"c": {N: &numStr2},
+					"d": {BOOL: &boolVal},
+				},
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+
+	tableName := "testtable"
+	cols := []string{"a", "b", "c", "d"}
+	conv := buildConv(
+		ddl.CreateTable{
+			Name:     tableName,
+			ColNames: cols,
+			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}},
+			},
+			Pks: []ddl.IndexKey{{Col: "a"}},
+		},
+		schema.Table{
+			Name:     tableName,
+			ColNames: cols,
+			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}},
+			},
+			PrimaryKeys: []schema.Key{{Column: "a"}},
+		},
+	)
+	conv.SetDataMode()
+	var rows []spannerData
+	conv.SetDataSink(
+		func(table string, cols []string, vals []interface{}) {
+			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
+		})
+	common.ProcessData(conv, InfoSchemaImpl{client, 10})
+	assert.Equal(t,
+		[]spannerData{
+			{
+				table: tableName,
+				cols:  cols,
+				vals:  []interface{}{"str-1", *numVal1, "12.34", true},
+			},
+		},
+		rows,
+	)
+}
+
 func TestInferDataTypes(t *testing.T) {
-	dySchema := schema.Table{Name: "test"}
 	stats := map[string]map[string]int64{
 		"all_rows_not_null": {
 			"Number": 1000,
@@ -317,14 +413,15 @@ func TestInferDataTypes(t *testing.T) {
 		},
 		"empty_stats": {},
 	}
-	inferDataTypes(stats, 1000, &dySchema)
+	colDefs, colNames, err := inferDataTypes(stats, 1000, make([]string, 0))
+	assert.Nil(t, err)
 	expectColNames := []string{
 		"all_rows_not_null", "err_row", "err_null_row", "enough_null_row",
 		"not_conflict_row", "conflict_row", "equal_conflict_rows",
 		"not_conflict_row_with_noise", "conflict_row_with_noise",
 		"equal_conflict_row_with_noise",
 	}
-	assert.ElementsMatch(t, expectColNames, dySchema.ColNames)
+	assert.ElementsMatch(t, expectColNames, colNames)
 	assert.Equal(t, map[string]schema.Column{
 		"all_rows_not_null":             {Name: "all_rows_not_null", Type: schema.Type{Name: "Number"}, NotNull: true},
 		"err_row":                       {Name: "err_row", Type: schema.Type{Name: "Number"}, NotNull: true},
@@ -336,7 +433,7 @@ func TestInferDataTypes(t *testing.T) {
 		"not_conflict_row_with_noise":   {Name: "not_conflict_row_with_noise", Type: schema.Type{Name: "Number"}, NotNull: false},
 		"conflict_row_with_noise":       {Name: "conflict_row_with_noise", Type: schema.Type{Name: "String"}, NotNull: false},
 		"equal_conflict_row_with_noise": {Name: "equal_conflict_row_with_noise", Type: schema.Type{Name: "String"}, NotNull: false},
-	}, dySchema.ColDefs)
+	}, colDefs)
 }
 
 func TestScanSampleData(t *testing.T) {
@@ -391,7 +488,7 @@ func TestScanSampleData(t *testing.T) {
 	assert.Equal(t, expectedStats, stats)
 }
 
-func TestParseIndexes(t *testing.T) {
+func TestInfoSchemaImpl_GetIndexes(t *testing.T) {
 	tableName := "test"
 	attrNameA := "a"
 	attrNameB := "b"
@@ -433,21 +530,73 @@ func TestParseIndexes(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 	}
 
-	dySchema := schema.Table{Name: "test"}
-
-	err := analyzeMetadata(client, &dySchema)
+	dySchema := common.SchemaAndName{Name: "test"}
+	conv := internal.MakeConv()
+	isi := InfoSchemaImpl{client, 10}
+	indexes, err := isi.GetIndexes(conv, dySchema)
 	assert.Nil(t, err)
 
-	pKeys := []schema.Key{{Column: "a"}, {Column: "b"}}
-	assert.Equal(t, pKeys, dySchema.PrimaryKeys)
 	secIndexes := []schema.Index{
 		{Name: "secondary_index_c", Keys: []schema.Key{{Column: "c"}}},
 		{Name: "secondary_index_d", Keys: []schema.Key{{Column: "d"}}},
 	}
-	assert.Equal(t, secIndexes, dySchema.Indexes)
+	assert.Equal(t, secIndexes, indexes)
 }
 
-func TestListTables(t *testing.T) {
+func TestInfoSchemaImpl_GetConstraints(t *testing.T) {
+	tableName := "test"
+	attrNameA := "a"
+	attrNameB := "b"
+	attrNameC := "c"
+	attrNameD := "d"
+	hashKeyType := "HASH"
+	sortKeyType := "RANGE"
+	globalIndexName := "secondary_index_c"
+	localIndexName := "secondary_index_d"
+	describeTableOutputs := []dynamodb.DescribeTableOutput{
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableName,
+				KeySchema: []*dynamodb.KeySchemaElement{
+					{AttributeName: &attrNameA, KeyType: &hashKeyType},
+					{AttributeName: &attrNameB, KeyType: &sortKeyType},
+				},
+				GlobalSecondaryIndexes: []*dynamodb.GlobalSecondaryIndexDescription{
+					{
+						IndexName: &globalIndexName,
+						KeySchema: []*dynamodb.KeySchemaElement{
+							{AttributeName: &attrNameC, KeyType: &hashKeyType},
+						},
+					},
+				},
+				LocalSecondaryIndexes: []*dynamodb.LocalSecondaryIndexDescription{
+					{
+						IndexName: &localIndexName,
+						KeySchema: []*dynamodb.KeySchemaElement{
+							{AttributeName: &attrNameD, KeyType: &hashKeyType},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		describeTableOutputs: describeTableOutputs,
+	}
+
+	dySchema := common.SchemaAndName{Name: "test"}
+	conv := internal.MakeConv()
+	isi := InfoSchemaImpl{client, 10}
+	primaryKeys, constraints, err := isi.GetConstraints(conv, dySchema)
+	assert.Nil(t, err)
+
+	pKeys := []string{"a", "b"}
+	assert.Equal(t, pKeys, primaryKeys)
+	assert.Empty(t, constraints)
+}
+
+func TestInfoSchemaImpl_GetTables(t *testing.T) {
 	tableNameA := "table-a"
 	tableNameB := "table-b"
 
@@ -459,10 +608,217 @@ func TestListTables(t *testing.T) {
 	client := &mockDynamoClient{
 		listTableOutputs: listTableOutputs,
 	}
+	isi := InfoSchemaImpl{client, 10}
+	tables, err := isi.GetTables()
+	assert.Nil(t, err)
+	assert.Equal(t, []common.SchemaAndName{{"", "table-a"}, {"", "table-b"}}, tables)
+}
+
+func TestInfoSchemaImpl_GetTableName(t *testing.T) {
+	tableNameA := "table-a"
 
-	tables, err := listTables(client)
+	client := &mockDynamoClient{}
+	isi := InfoSchemaImpl{client, 10}
+	table := isi.GetTableName("", tableNameA)
+	assert.Equal(t, tableNameA, table)
+}
+
+func TestInfoSchemaImpl_GetColumns(t *testing.T) {
+	strA := "str-1"
+	strB := "str-2"
+	numStr := "10"
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+				},
+				{
+					"a": {N: &numStr},
+				},
+			},
+			LastEvaluatedKey: map[string]*dynamodb.AttributeValue{
+				"a": {S: &strA},
+			},
+		},
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strB},
+					"b": {N: &numStr},
+				},
+				{
+					// This will not be scaned due to the sample size.
+					"a": {N: &numStr},
+					"b": {S: &strB},
+				},
+			},
+		},
+	}
+
+	conv := internal.MakeConv()
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+	dySchema := common.SchemaAndName{Name: "test"}
+
+	isi := InfoSchemaImpl{client, 10}
+
+	colDefs, colNames, err := isi.GetColumns(conv, dySchema, nil, nil)
+	assert.Nil(t, err)
+	expectColNames := []string{
+		"a", "b",
+	}
+	assert.ElementsMatch(t, expectColNames, colNames)
+	assert.Equal(t, map[string]schema.Column{
+		"a": {Name: "a", Type: schema.Type{Name: "String", Mods: []int64(nil), ArrayBounds: []int64(nil)}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}},
+		"b": {Name: "b", Type: schema.Type{Name: "String", Mods: []int64(nil), ArrayBounds: []int64(nil)}, NotNull: false, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}}},
+		colDefs)
+}
+
+func TestInfoSchemaImpl_GetForeignKeys(t *testing.T) {
+	dySchema := common.SchemaAndName{Name: "test"}
+	conv := internal.MakeConv()
+	client := &mockDynamoClient{}
+	isi := InfoSchemaImpl{client, 10}
+	fk, err := isi.GetForeignKeys(conv, dySchema)
+	assert.Nil(t, err)
+	assert.Nil(t, fk)
+}
+
+func TestInfoSchemaImpl_GetRowCount(t *testing.T) {
+	tableNameA := "test_a"
+	tableItemCountA := int64(10)
+
+	describeTableOutputs := []dynamodb.DescribeTableOutput{
+		{
+			Table: &dynamodb.TableDescription{
+				TableName: &tableNameA,
+				ItemCount: &tableItemCountA,
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		describeTableOutputs: describeTableOutputs,
+	}
+
+	isi := InfoSchemaImpl{client, 10}
+	dySchema := common.SchemaAndName{Name: tableNameA}
+
+	rowCount, err := isi.GetRowCount(dySchema)
 	assert.Nil(t, err)
-	assert.Equal(t, []string{"table-a", "table-b"}, tables)
+	assert.Equal(t, tableItemCountA, rowCount)
+}
+
+func TestInfoSchemaImpl_GetRowsFromTable(t *testing.T) {
+	strA := "str-1"
+	numStr1 := "10.1"
+	numStr2 := "12.34"
+
+	boolVal := true
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+					"b": {N: &numStr1},
+					"c": {N: &numStr2},
+					"d": {BOOL: &boolVal},
+				},
+			},
+		},
+	}
+
+	conv := internal.MakeConv()
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+	tableName := "testtable"
+	isi := InfoSchemaImpl{client, 10}
+
+	rows, err := isi.GetRowsFromTable(conv, tableName)
+	assert.Nil(t, err)
+	assert.Equal(t, []map[string]*dynamodb.AttributeValue{{
+		"a": {S: &strA},
+		"b": {N: &numStr1},
+		"c": {N: &numStr2},
+		"d": {BOOL: &boolVal}}},
+		rows,
+	)
+}
+
+func TestInfoSchemaImpl_ProcessData(t *testing.T) {
+	strA := "str-1"
+	numStr1 := "10.1"
+	numStr2 := "12.34"
+	numVal1 := big.NewRat(101, 10)
+
+	boolVal := true
+	scanOutputs := []dynamodb.ScanOutput{
+		{
+			Items: []map[string]*dynamodb.AttributeValue{
+				{
+					"a": {S: &strA},
+					"b": {N: &numStr1},
+					"c": {N: &numStr2},
+					"d": {BOOL: &boolVal},
+				},
+			},
+		},
+	}
+
+	client := &mockDynamoClient{
+		scanOutputs: scanOutputs,
+	}
+	isi := InfoSchemaImpl{client, 10}
+
+	tableName := "testtable"
+	cols := []string{"a", "b", "c", "d"}
+	spSchema := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: cols,
+		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}},
+		},
+		Pks: []ddl.IndexKey{{Col: "a"}},
+	}
+	conv := buildConv(
+		spSchema,
+		schema.Table{
+			Name:     tableName,
+			ColNames: cols,
+			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}},
+			},
+			PrimaryKeys: []schema.Key{{Column: "a"}},
+		},
+	)
+
+	var rows []spannerData
+	conv.SetDataSink(
+		func(table string, cols []string, vals []interface{}) {
+			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
+		})
+	err := isi.ProcessData(conv, tableName, conv.SrcSchema[tableName], tableName,
+		cols, spSchema)
+	assert.Nil(t, err)
+	assert.Equal(t,
+		[]spannerData{
+			{
+				table: tableName,
+				cols:  cols,
+				vals:  []interface{}{"str-1", *numVal1, "12.34", true},
+			},
+		},
+		rows,
+	)
 }
 
 func stripSchemaComments(spSchema map[string]ddl.CreateTable) map[string]ddl.CreateTable {
@@ -550,7 +906,7 @@ func TestSetRowStats(t *testing.T) {
 		describeTableOutputs: describeTableOutputs,
 	}
 
-	SetRowStats(conv, client)
+	common.SetRowStats(conv, InfoSchemaImpl{client, 10})
 
 	assert.Equal(t, tableItemCountA, conv.Stats.Rows[tableNameA])
 	assert.Equal(t, tableItemCountB, conv.Stats.Rows[tableNameB])
diff --git a/sources/mysql/data_test.go b/sources/mysql/data_test.go
index 930278b..ebe3937 100644
--- a/sources/mysql/data_test.go
+++ b/sources/mysql/data_test.go
@@ -88,7 +88,7 @@ func TestConvertData(t *testing.T) {
 		{"string", ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, "", "eh", "eh"},
 		{"datetime", ddl.Type{Name: ddl.Timestamp}, "datetime", "2019-10-29 05:30:00", getTimeWithoutTimezone(t, "2019-10-29 05:30:00")},
 		{"timestamp", ddl.Type{Name: ddl.Timestamp}, "timestamp", "2019-10-29 05:30:00", getTime(t, "2019-10-29T05:30:00+05:30")},
-		{"json", ddl.Type{Name: ddl.Json}, "", "{\"key1\": \"value1\"}", "{\"key1\": \"value1\"}"},
+		{"json", ddl.Type{Name: ddl.JSON}, "", "{\"key1\": \"value1\"}", "{\"key1\": \"value1\"}"},
 		{"string array(set)", ddl.Type{Name: ddl.String, Len: ddl.MaxLength, IsArray: true}, "", "1,Travel,3,Dance", []spanner.NullString{
 			spanner.NullString{StringVal: "1", Valid: true},
 			spanner.NullString{StringVal: "Travel", Valid: true},
@@ -320,12 +320,6 @@ func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *internal.Conv {
 	conv := internal.MakeConv()
 	conv.SpSchema[spTable.Name] = spTable
 	conv.SrcSchema[srcTable.Name] = srcTable
-	conv.ToSource[spTable.Name] = internal.NameAndCols{Name: srcTable.Name, Cols: make(map[string]string)}
-	conv.ToSpanner[srcTable.Name] = internal.NameAndCols{Name: spTable.Name, Cols: make(map[string]string)}
-	for i := range spTable.ColNames {
-		conv.ToSource[spTable.Name].Cols[spTable.ColNames[i]] = srcTable.ColNames[i]
-		conv.ToSpanner[srcTable.Name].Cols[srcTable.ColNames[i]] = spTable.ColNames[i]
-	}
 	return conv
 }
 
diff --git a/sources/mysql/infoschema_test.go b/sources/mysql/infoschema_test.go
index 0c56fe2..0f1c841 100644
--- a/sources/mysql/infoschema_test.go
+++ b/sources/mysql/infoschema_test.go
@@ -34,7 +34,7 @@ type mockSpec struct {
 	rows  [][]driver.Value // Set of rows returned.
 }
 
-func TestProcessInfoSchemaMYSQL(t *testing.T) {
+func TestProcessSchemaMYSQL(t *testing.T) {
 	ms := []mockSpec{
 
 		{
@@ -47,14 +47,6 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"product"},
 				{"test"},
 				{"test_ref"}},
-		}, {
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "user"},
-			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
-			rows: [][]driver.Value{
-				{"user_id", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"name", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"ref", "bigint", "bigint", "NO", nil, nil, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "user"},
@@ -73,16 +65,16 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
 			args:  []driver.Value{"test", "user"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
-		},
-		{
+		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "cart"},
+			args:  []driver.Value{"test", "user"},
 			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
 			rows: [][]driver.Value{
-				{"productid", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"userid", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"quantity", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
-		}, {
+				{"user_id", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"name", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"ref", "bigint", "bigint", "NO", nil, nil, nil, nil, nil}},
+		},
+		{
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "cart"},
 			cols:  []string{"column_name", "constraint_type"},
@@ -108,11 +100,12 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"index3", "userid", 2, "D", "0"}},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "product"},
+			args:  []driver.Value{"test", "cart"},
 			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
 			rows: [][]driver.Value{
-				{"product_id", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"product_name", "text", "text", "NO", nil, nil, nil, nil, nil}},
+				{"productid", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"userid", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"quantity", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "product"},
@@ -127,6 +120,28 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
 			args:  []driver.Value{"test", "product"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"test", "product"},
+			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
+			rows: [][]driver.Value{
+				{"product_id", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"product_name", "text", "text", "NO", nil, nil, nil, nil, nil}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}, {"id", "FOREIGN KEY"}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
+			rows: [][]driver.Value{{"test_ref", "id", "ref_id", "fk_test4"},
+				{"test_ref", "txt", "ref_txt", "fk_test4"}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"test", "test"},
@@ -154,18 +169,18 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"vc6", "varchar", "varchar(6)", "YES", nil, 6, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test"},
+			args:  []driver.Value{"test", "test_ref"},
 			cols:  []string{"column_name", "constraint_type"},
-			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}, {"id", "FOREIGN KEY"}},
+			rows: [][]driver.Value{
+				{"ref_id", "PRIMARY KEY"},
+				{"ref_txt", "PRIMARY KEY"}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test"},
+			args:  []driver.Value{"test", "test_ref"},
 			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
-			rows: [][]driver.Value{{"test_ref", "id", "ref_id", "fk_test4"},
-				{"test_ref", "txt", "ref_txt", "fk_test4"}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
-			args:  []driver.Value{"test", "test"},
+			args:  []driver.Value{"test", "test_ref"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
@@ -175,26 +190,11 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"ref_id", "bigint", "bigint", "NO", nil, nil, 64, 0, nil},
 				{"ref_txt", "text", "text", "NO", nil, nil, nil, nil, nil},
 				{"abc", "text", "text", "NO", nil, nil, nil, nil, nil}},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test_ref"},
-			cols:  []string{"column_name", "constraint_type"},
-			rows: [][]driver.Value{
-				{"ref_id", "PRIMARY KEY"},
-				{"ref_txt", "PRIMARY KEY"}},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test_ref"},
-			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
-			args:  []driver.Value{"test", "test_ref"},
-			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		},
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{"test"})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"user": ddl.CreateTable{
@@ -280,14 +280,9 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 	assert.Equal(t, int64(0), conv.Unexpecteds())
 }
 
-func TestProcessSQLData(t *testing.T) {
+func TestProcessData(t *testing.T) {
 	ms := []mockSpec{
 		{
-			query: "SELECT table_name FROM information_schema.tables where table_type = 'BASE TABLE' and (.+)",
-			args:  []driver.Value{"test"},
-			cols:  []string{"table_name"},
-			rows:  [][]driver.Value{{"te st"}},
-		}, {
 			query: "SELECT (.+) FROM `test`.`te st`",
 			cols:  []string{"a a", " b", " c "},
 			rows: [][]driver.Value{
@@ -308,6 +303,7 @@ func TestProcessSQLData(t *testing.T) {
 			}},
 		schema.Table{
 			Name:     "te st",
+			Schema:   "test",
 			ColNames: []string{"a_a", "_b", "_c_"},
 			ColDefs: map[string]schema.Column{
 				"a a": schema.Column{Name: "a a", Type: schema.Type{Name: "float"}},
@@ -321,7 +317,7 @@ func TestProcessSQLData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{"test"})
+	common.ProcessData(conv, InfoSchemaImpl{"test", db})
 	assert.Equal(t,
 		[]spannerData{
 			spannerData{table: "te_st", cols: []string{"a_a", "Ab", "Ac_"}, vals: []interface{}{float64(42.3), int64(3), "cat"}},
@@ -333,7 +329,7 @@ func TestProcessSQLData(t *testing.T) {
 	assert.Equal(t, int64(1), conv.Unexpecteds()) // Bad row generates an entry in unexpected.
 }
 
-func TestProcessSQLData_MultiCol(t *testing.T) {
+func TestProcessData_MultiCol(t *testing.T) {
 	// Tests multi-column behavior of ProcessSQLData (including
 	// handling of null columns and synthetic keys). Also tests
 	// the combination of ProcessInfoSchema and ProcessSQLData
@@ -346,15 +342,6 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 			cols:  []string{"table_name"},
 			rows:  [][]driver.Value{{"test"}},
 		}, {
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "test"},
-			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
-			rows: [][]driver.Value{
-				{"a", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"b", "double", "double", "YES", nil, nil, 53, nil, nil},
-				{"c", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
-		},
-		{
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "test"},
 			cols:  []string{"column_name", "constraint_type"},
@@ -367,17 +354,16 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
 			args:  []driver.Value{"test", "test"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
+			rows: [][]driver.Value{
+				{"a", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"b", "double", "double", "YES", nil, nil, 53, nil, nil},
+				{"c", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
 		},
-		// Note: go-sqlmock mocks specify an ordered sequence
-		// of queries and results.  This (repeated) entry is
-		// needed because ProcessSQLData (redundantly) gets
-		// the set of tables via a SQL query.
 		{
-			query: "SELECT table_name FROM information_schema.tables where table_type = 'BASE TABLE' and (.+)",
-			args:  []driver.Value{"test"},
-			cols:  []string{"table_name"},
-			rows:  [][]driver.Value{{"test"}},
-		}, {
 			query: "SELECT (.+) FROM `test`.`test`",
 			cols:  []string{"a", "b", "c"},
 			rows: [][]driver.Value{
@@ -387,7 +373,7 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{"test"})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"test": ddl.CreateTable{
@@ -411,7 +397,7 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{"test"})
+	common.ProcessData(conv, InfoSchemaImpl{"test", db})
 	assert.Equal(t, []spannerData{
 		{table: "test", cols: []string{"a", "b", "synth_id"}, vals: []interface{}{"cat", float64(42.3), int64(0)}},
 		{table: "test", cols: []string{"a", "c", "synth_id"}, vals: []interface{}{"dog", int64(22), int64(-9223372036854775808)}}},
@@ -439,7 +425,7 @@ func TestSetRowStats(t *testing.T) {
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
 	conv.SetDataMode()
-	common.SetRowStats(conv, db, InfoSchemaImpl{"test"})
+	common.SetRowStats(conv, InfoSchemaImpl{"test", db})
 	assert.Equal(t, int64(5), conv.Stats.Rows["test1"])
 	assert.Equal(t, int64(142), conv.Stats.Rows["test2"])
 	assert.Equal(t, int64(0), conv.Unexpecteds())
diff --git a/sources/mysql/toddl_test.go b/sources/mysql/toddl_test.go
index 171fcd7..c849d57 100644
--- a/sources/mysql/toddl_test.go
+++ b/sources/mysql/toddl_test.go
@@ -80,7 +80,7 @@ func TestToSpannerType(t *testing.T) {
 			"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.Json}},
+			"g": ddl.ColumnDef{Name: "g", T: ddl.Type{Name: ddl.JSON}},
 		},
 		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/postgres/data_test.go b/sources/postgres/data_test.go
index aa05951..6c05ce0 100644
--- a/sources/postgres/data_test.go
+++ b/sources/postgres/data_test.go
@@ -280,12 +280,6 @@ func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *internal.Conv {
 	conv := internal.MakeConv()
 	conv.SpSchema[spTable.Name] = spTable
 	conv.SrcSchema[srcTable.Name] = srcTable
-	conv.ToSource[spTable.Name] = internal.NameAndCols{Name: srcTable.Name, Cols: make(map[string]string)}
-	conv.ToSpanner[srcTable.Name] = internal.NameAndCols{Name: spTable.Name, Cols: make(map[string]string)}
-	for i := range spTable.ColNames {
-		conv.ToSource[spTable.Name].Cols[spTable.ColNames[i]] = srcTable.ColNames[i]
-		conv.ToSpanner[srcTable.Name].Cols[srcTable.ColNames[i]] = spTable.ColNames[i]
-	}
 	return conv
 }
 
diff --git a/sources/postgres/infoschema_test.go b/sources/postgres/infoschema_test.go
index b623e7a..fe61a5c 100644
--- a/sources/postgres/infoschema_test.go
+++ b/sources/postgres/infoschema_test.go
@@ -37,7 +37,7 @@ type mockSpec struct {
 	rows  [][]driver.Value // Set of rows returned.
 }
 
-func TestProcessInfoSchema(t *testing.T) {
+func TestProcessSchema(t *testing.T) {
 	ms := []mockSpec{
 		{
 			query: "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'",
@@ -48,15 +48,6 @@ func TestProcessInfoSchema(t *testing.T) {
 				{"public", "product"},
 				{"public", "test"},
 				{"public", "test_ref"}},
-		},
-		{
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "user"},
-			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
-			rows: [][]driver.Value{
-				{"user_id", "text", nil, "NO", nil, nil, nil, nil},
-				{"name", "text", nil, "NO", nil, nil, nil, nil},
-				{"ref", "bigint", nil, "YES", nil, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "user"},
@@ -77,12 +68,12 @@ func TestProcessInfoSchema(t *testing.T) {
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "cart"},
+			args:  []driver.Value{"public", "user"},
 			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
 			rows: [][]driver.Value{
-				{"productid", "text", nil, "NO", nil, nil, nil, nil},
-				{"userid", "text", nil, "NO", nil, nil, nil, nil},
-				{"quantity", "bigint", nil, "YES", nil, nil, 64, 0}},
+				{"user_id", "text", nil, "NO", nil, nil, nil, nil},
+				{"name", "text", nil, "NO", nil, nil, nil, nil},
+				{"ref", "bigint", nil, "YES", nil, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "cart"},
@@ -109,11 +100,12 @@ func TestProcessInfoSchema(t *testing.T) {
 			},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "product"},
+			args:  []driver.Value{"public", "cart"},
 			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
 			rows: [][]driver.Value{
-				{"product_id", "text", nil, "NO", nil, nil, nil, nil},
-				{"product_name", "text", nil, "NO", nil, nil, nil, nil}},
+				{"productid", "text", nil, "NO", nil, nil, nil, nil},
+				{"userid", "text", nil, "NO", nil, nil, nil, nil},
+				{"quantity", "bigint", nil, "YES", nil, nil, 64, 0}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "product"},
@@ -128,6 +120,28 @@ func TestProcessInfoSchema(t *testing.T) {
 			query: "SELECT (.+) FROM pg_index (.+)",
 			args:  []driver.Value{"public", "product"},
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"public", "product"},
+			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"product_id", "text", nil, "NO", nil, nil, nil, nil},
+				{"product_name", "text", nil, "NO", nil, nil, nil, nil}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}},
+		}, {
+			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+			rows: [][]driver.Value{{"public", "test_ref", "id", "ref_id", "fk_test4"},
+				{"public", "test_ref", "txt", "ref_txt", "fk_test4"}},
+		}, {
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"public", "test"},
@@ -156,18 +170,18 @@ func TestProcessInfoSchema(t *testing.T) {
 				{"vc6", "character varying", nil, "YES", nil, 6, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"public", "test"},
+			args:  []driver.Value{"public", "test_ref"},
 			cols:  []string{"column_name", "constraint_type"},
-			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}},
+			rows: [][]driver.Value{
+				{"ref_id", "PRIMARY KEY"},
+				{"ref_txt", "PRIMARY KEY"}},
 		}, {
 			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
-			args:  []driver.Value{"public", "test"},
+			args:  []driver.Value{"public", "test_ref"},
 			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
-			rows: [][]driver.Value{{"public", "test_ref", "id", "ref_id", "fk_test4"},
-				{"public", "test_ref", "txt", "ref_txt", "fk_test4"}},
 		}, {
 			query: "SELECT (.+) FROM pg_index (.+)",
-			args:  []driver.Value{"public", "test"},
+			args:  []driver.Value{"public", "test_ref"},
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
@@ -177,26 +191,11 @@ func TestProcessInfoSchema(t *testing.T) {
 				{"ref_id", "bigint", nil, "NO", nil, nil, 64, 0},
 				{"ref_txt", "text", nil, "NO", nil, nil, nil, nil},
 				{"abc", "text", nil, "NO", nil, nil, nil, nil}},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"public", "test_ref"},
-			cols:  []string{"column_name", "constraint_type"},
-			rows: [][]driver.Value{
-				{"ref_id", "PRIMARY KEY"},
-				{"ref_txt", "PRIMARY KEY"}},
-		}, {
-			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
-			args:  []driver.Value{"public", "test_ref"},
-			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
-		}, {
-			query: "SELECT (.+) FROM pg_index (.+)",
-			args:  []driver.Value{"public", "test_ref"},
-			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		},
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{db})
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"user": ddl.CreateTable{
@@ -288,13 +287,9 @@ func TestProcessInfoSchema(t *testing.T) {
 // handling of bad rows and table and column renaming. The core data
 // conversion work of ProcessSqlData is done by ConvertData, which is
 // extensively is tested by TestConvertSqlRow.
-func TestProcessSqlData(t *testing.T) {
+func TestProcessData(t *testing.T) {
 	ms := []mockSpec{
 		{
-			query: "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'",
-			cols:  []string{"table_schema", "table_name"},
-			rows:  [][]driver.Value{{"public", "te st"}},
-		}, {
 			query: `SELECT [*] FROM "public"."te st"`, // query is a regexp!
 			cols:  []string{"a a", " b", " c "},
 			rows: [][]driver.Value{
@@ -315,6 +310,7 @@ func TestProcessSqlData(t *testing.T) {
 			}},
 		schema.Table{
 			Name:     "te st",
+			Schema:   "public",
 			ColNames: []string{"a_a", "_b", "_c_"},
 			ColDefs: map[string]schema.Column{
 				"a a": schema.Column{Name: "a a", Type: schema.Type{Name: "float4"}},
@@ -327,7 +323,7 @@ func TestProcessSqlData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{})
+	common.ProcessData(conv, InfoSchemaImpl{db})
 
 	assert.Equal(t,
 		[]spannerData{
@@ -423,15 +419,6 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 			cols:  []string{"table_schema", "table_name"},
 			rows:  [][]driver.Value{{"public", "test"}},
 		}, {
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "test"},
-			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
-			rows: [][]driver.Value{
-				{"a", "text", nil, "NO", nil, nil, nil, nil},
-				{"b", "double precision", nil, "YES", nil, nil, 53, nil},
-				{"c", "bigint", nil, "YES", nil, nil, 64, 0}},
-		},
-		{
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "test"},
 			cols:  []string{"column_name", "constraint_type"},
@@ -446,15 +433,14 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 			query: "SELECT (.+) FROM pg_index (.+)",
 			args:  []driver.Value{"public", "test"},
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
-		},
-		// Note: go-sqlmock mocks specify an ordered sequence
-		// of queries and results.  This (repeated) entry is
-		// needed because ProcessSqlData (redundantly) gets
-		// the set of tables via a SQL query.
-		{
-			query: "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'",
-			cols:  []string{"table_schema", "table_name"},
-			rows:  [][]driver.Value{{"public", "test"}},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"a", "text", nil, "NO", nil, nil, nil, nil},
+				{"b", "double precision", nil, "YES", nil, nil, 53, nil},
+				{"c", "bigint", nil, "YES", nil, nil, 64, 0}},
 		}, {
 			query: `SELECT [*] FROM "public"."test"`, // query is a regexp!
 			cols:  []string{"a", "b", "c"},
@@ -465,7 +451,7 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{db})
 	assert.Nil(t, err)
 	conv.SetDataMode()
 	var rows []spannerData
@@ -473,7 +459,7 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{})
+	common.ProcessData(conv, InfoSchemaImpl{db})
 	assert.Equal(t, []spannerData{
 		{table: "test", cols: []string{"a", "b", "synth_id"}, vals: []interface{}{"cat", float64(42.3), int64(0)}},
 		{table: "test", cols: []string{"a", "c", "synth_id"}, vals: []interface{}{"dog", int64(22), int64(-9223372036854775808)}}},
@@ -500,7 +486,7 @@ func TestSetRowStats(t *testing.T) {
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
 	conv.SetDataMode()
-	common.SetRowStats(conv, db, InfoSchemaImpl{})
+	common.SetRowStats(conv, InfoSchemaImpl{db})
 	assert.Equal(t, int64(5), conv.Stats.Rows["test1"])
 	assert.Equal(t, int64(142), conv.Stats.Rows["test2"])
 	assert.Equal(t, int64(0), conv.Unexpecteds())
diff --git a/sources/postgres/pgdump_test.go b/sources/postgres/pgdump_test.go
index d74f8a0..57c65ed 100644
--- a/sources/postgres/pgdump_test.go
+++ b/sources/postgres/pgdump_test.go
@@ -70,8 +70,8 @@ func TestProcessPgDump(t *testing.T) {
 		{"timestamptz(5)", ddl.Type{Name: ddl.Timestamp}},
 		{"varchar", ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
 		{"varchar(42)", ddl.Type{Name: ddl.String, Len: int64(42)}},
-		{"json", ddl.Type{Name: ddl.Json}},
-		{"jsonb", ddl.Type{Name: ddl.Json}},
+		{"json", ddl.Type{Name: ddl.JSON}},
+		{"jsonb", ddl.Type{Name: ddl.JSON}},
 	}
 	for _, tc := range scalarTests {
 		conv, _ := runProcessPgDump(fmt.Sprintf("CREATE TABLE t (a %s);", tc.ty))
@@ -1503,7 +1503,7 @@ func runProcessPgDump(s string) (*internal.Conv, []spannerData) {
 
 func runProcessPgDumpPGTarget(s string) (*internal.Conv, []spannerData) {
 	conv := internal.MakeConv()
-	conv.TargetDb = constants.TARGET_EXPERIMENTAL_POSTGRES
+	conv.TargetDb = constants.TargetExperimentalPostgres
 	conv.SetLocation(time.UTC)
 	conv.SetSchemaMode()
 	pgDump := DbDumpImpl{}
diff --git a/sources/postgres/toddl_test.go b/sources/postgres/toddl_test.go
index 0552f6c..0cc9bce 100644
--- a/sources/postgres/toddl_test.go
+++ b/sources/postgres/toddl_test.go
@@ -103,7 +103,7 @@ func TestToSpannerType(t *testing.T) {
 func TestToExperimentalSpannerType(t *testing.T) {
 	conv := internal.MakeConv()
 	conv.SetSchemaMode()
-	conv.TargetDb = constants.TARGET_EXPERIMENTAL_POSTGRES
+	conv.TargetDb = constants.TargetExperimentalPostgres
 	name := "test"
 	srcSchema := schema.Table{
 		Name:     name,
diff --git a/spanner/ddl/ast_test.go b/spanner/ddl/ast_test.go
index f605a05..cfaafce 100644
--- a/spanner/ddl/ast_test.go
+++ b/spanner/ddl/ast_test.go
@@ -91,7 +91,7 @@ func TestPrintColumnDefPG(t *testing.T) {
 		{in: ColumnDef{Name: "col1", T: Type{Name: Int64}}, protectIds: true, expected: "`col1` INT8"},
 	}
 	for _, tc := range tests {
-		s, _ := tc.in.PrintColumnDef(Config{ProtectIds: tc.protectIds, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES})
+		s, _ := tc.in.PrintColumnDef(Config{ProtectIds: tc.protectIds, TargetDb: constants.TargetExperimentalPostgres})
 		assert.Equal(t, tc.expected, s)
 	}
 }
@@ -108,7 +108,7 @@ func TestPrintIndexKey(t *testing.T) {
 	}
 	for _, tc := range tests {
 		assert.Equal(t, tc.expected, tc.in.PrintIndexKey(Config{ProtectIds: tc.protectIds}))
-		assert.Equal(t, tc.expected, tc.in.PrintIndexKey(Config{ProtectIds: tc.protectIds, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES}))
+		assert.Equal(t, tc.expected, tc.in.PrintIndexKey(Config{ProtectIds: tc.protectIds, TargetDb: constants.TargetExperimentalPostgres}))
 	}
 }
 
@@ -246,7 +246,7 @@ func TestPrintCreateTablePG(t *testing.T) {
 		},
 	}
 	for _, tc := range tests {
-		assert.Equal(t, tc.expected, tc.ct.PrintCreateTable(Config{ProtectIds: tc.protectIds, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES}))
+		assert.Equal(t, tc.expected, tc.ct.PrintCreateTable(Config{ProtectIds: tc.protectIds, TargetDb: constants.TargetExperimentalPostgres}))
 	}
 }
 
@@ -276,7 +276,7 @@ func TestPrintCreateIndex(t *testing.T) {
 	}
 	for _, tc := range tests {
 		assert.Equal(t, tc.expected, tc.index.PrintCreateIndex(Config{ProtectIds: tc.protectIds}))
-		assert.Equal(t, tc.expected, tc.index.PrintCreateIndex(Config{ProtectIds: tc.protectIds, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES}))
+		assert.Equal(t, tc.expected, tc.index.PrintCreateIndex(Config{ProtectIds: tc.protectIds, TargetDb: constants.TargetExperimentalPostgres}))
 	}
 }
 
@@ -307,7 +307,7 @@ func TestPrintForeignKey(t *testing.T) {
 	}
 	for _, tc := range tests {
 		assert.Equal(t, tc.expected, tc.fk.PrintForeignKey(Config{ProtectIds: tc.protectIds}))
-		assert.Equal(t, tc.expected, tc.fk.PrintForeignKey(Config{ProtectIds: tc.protectIds, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES}))
+		assert.Equal(t, tc.expected, tc.fk.PrintForeignKey(Config{ProtectIds: tc.protectIds, TargetDb: constants.TargetExperimentalPostgres}))
 	}
 }
 
@@ -339,7 +339,7 @@ func TestPrintForeignKeyAlterTable(t *testing.T) {
 	}
 	for _, tc := range tests {
 		assert.Equal(t, tc.expected, tc.fk.PrintForeignKeyAlterTable(Config{ProtectIds: tc.protectIds}, tc.table))
-		assert.Equal(t, tc.expected, tc.fk.PrintForeignKeyAlterTable(Config{ProtectIds: tc.protectIds, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES}, tc.table))
+		assert.Equal(t, tc.expected, tc.fk.PrintForeignKeyAlterTable(Config{ProtectIds: tc.protectIds, TargetDb: constants.TargetExperimentalPostgres}, tc.table))
 	}
 }
 
@@ -473,7 +473,7 @@ func TestGetPGDDL(t *testing.T) {
 		Fks:    []Foreignkey{{Name: "fk3", Columns: []string{"c"}, ReferTable: "ref_table3", ReferColumns: []string{"ref_c"}}},
 		Parent: "table1",
 	}
-	tablesOnly := s.GetDDL(Config{Tables: true, ForeignKeys: false, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES})
+	tablesOnly := s.GetDDL(Config{Tables: true, ForeignKeys: false, TargetDb: constants.TargetExperimentalPostgres})
 	e := []string{
 		"CREATE TABLE table1 (\n" +
 			"	a INT8,\n" +
@@ -497,7 +497,7 @@ func TestGetPGDDL(t *testing.T) {
 	}
 	assert.ElementsMatch(t, e, tablesOnly)
 
-	fksOnly := s.GetDDL(Config{Tables: false, ForeignKeys: true, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES})
+	fksOnly := s.GetDDL(Config{Tables: false, ForeignKeys: true, TargetDb: constants.TargetExperimentalPostgres})
 	e2 := []string{
 		"ALTER TABLE table1 ADD CONSTRAINT fk1 FOREIGN KEY (b) REFERENCES ref_table1 (ref_b)",
 		"ALTER TABLE table2 ADD CONSTRAINT fk2 FOREIGN KEY (b, c) REFERENCES ref_table2 (ref_b, ref_c)",
@@ -505,7 +505,7 @@ func TestGetPGDDL(t *testing.T) {
 	}
 	assert.ElementsMatch(t, e2, fksOnly)
 
-	tablesAndFks := s.GetDDL(Config{Tables: true, ForeignKeys: true, TargetDb: constants.TARGET_EXPERIMENTAL_POSTGRES})
+	tablesAndFks := s.GetDDL(Config{Tables: true, ForeignKeys: true, TargetDb: constants.TargetExperimentalPostgres})
 	e3 := []string{
 		"CREATE TABLE table1 (\n" +
 			"	a INT8,\n" +
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index 6af891a..aeef596 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -126,20 +126,20 @@ func TestIntegration_MYSQLDUMP_Command(t *testing.T) {
 	checkResults(t, dbURI)
 }
 
-func TestIntegration_MySQL_EvalSubcommand(t *testing.T) {
+func TestIntegration_MYSQL_SchemaAndDataSubcommand(t *testing.T) {
 	onlyRunForEmulatorTest(t)
 	t.Parallel()
 
 	tmpdir := prepareIntegrationTest(t)
 	defer os.RemoveAll(tmpdir)
 
-	dbName := "mysql-dc-eval-src-params"
+	dbName := "mysql-dc-schema-and-data"
 	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")
 	envVars := common.ClearEnvVariables([]string{"MYSQLHOST", "MYSQLUSER", "MYSQLDATABASE", "MYSQLPWD"})
-	args := fmt.Sprintf("eval -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,db_name=%s,password=%s' -target-profile='instance=%s,dbname=%s'", constants.MYSQL, filePrefix, host, user, db_name, password, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	common.RestoreEnvVariables(envVars)
 	if err != nil {
@@ -246,8 +246,8 @@ func runDataSubcommand(t *testing.T, dbName, dbURI, filePrefix, sessionFile, dum
 	}
 }
 
-func runEvalSubcommand(t *testing.T, dbName, dbURI, filePrefix, dumpFilePath string) {
-	args := fmt.Sprintf("eval -source=mysql -prefix %s -target-profile='instance=%s,dbname=%s' < %s", filePrefix, instanceID, dbName, dumpFilePath)
+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)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -294,17 +294,17 @@ func TestIntegration_MySQLDUMP_DataSubcommand(t *testing.T) {
 	checkResults(t, dbURI)
 }
 
-func TestIntegration_MySQLDUMP_EvalSubcommand(t *testing.T) {
+func TestIntegration_MySQLDUMP_SchemaAndDataSubcommand(t *testing.T) {
 	onlyRunForEmulatorTest(t)
 	tmpdir := prepareIntegrationTest(t)
 	defer os.RemoveAll(tmpdir)
 
-	dbName := "test-eval-subcommand"
+	dbName := "test-schema-and-data"
 	dumpFilePath := "../../test_data/mysqldump.test.out"
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	runEvalSubcommand(t, dbName, dbURI, filePrefix, dumpFilePath)
+	runSchemaAndDataSubcommand(t, dbName, dbURI, filePrefix, dumpFilePath)
 	defer dropDatabase(t, dbURI)
 	checkResults(t, dbURI)
 }
diff --git a/testing/postgres/integration_test.go b/testing/postgres/integration_test.go
index b859dcb..7673246 100644
--- a/testing/postgres/integration_test.go
+++ b/testing/postgres/integration_test.go
@@ -129,7 +129,7 @@ func TestIntegration_PGDUMP_Command(t *testing.T) {
 	checkResults(t, dbURI)
 }
 
-func TestIntegration_PGDUMP_EvalSubcommand(t *testing.T) {
+func TestIntegration_PGDUMP_SchemaAndDataSubcommand(t *testing.T) {
 	t.Parallel()
 
 	tmpdir := prepareIntegrationTest(t)
@@ -142,7 +142,7 @@ func TestIntegration_PGDUMP_EvalSubcommand(t *testing.T) {
 	dataFilepath := "../../test_data/pg_dump.test.out"
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	args := fmt.Sprintf("eval -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)
@@ -191,7 +191,7 @@ func TestIntegration_POSTGRES_Command(t *testing.T) {
 	checkResults(t, dbURI)
 }
 
-func TestIntegration_POSTGRES_EvalSubcommand(t *testing.T) {
+func TestIntegration_POSTGRES_SchemaAndDataSubcommand(t *testing.T) {
 	onlyRunForEmulatorTest(t)
 	t.Parallel()
 
@@ -203,7 +203,7 @@ func TestIntegration_POSTGRES_EvalSubcommand(t *testing.T) {
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	args := fmt.Sprintf("eval -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)