Repository: cloudspannerecosystem/harbourbridge
Pull request #256 (2022-01-18 14:55:21)

CSV 2
This adds the following features on top of csv:

- No longer provide data types in manifest file. HB reads spanner infoschema to fetch the data types.
- Support for PG spanner
- Interleaved table support
- GCS file support
- Array support added



Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index a4ea0bb..0574a45 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -7,7 +7,7 @@ orbs:
 jobs:
   build_and_test:
     docker:
-      - image: circleci/golang:1.13
+      - image: circleci/golang:1.16
 
     working_directory: /go/src/github.com/cloudspannerecosystem/harbourbridge
 
diff --git a/cmd/cmd.go b/cmd/cmd.go
index a3509c1..287cb76 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -18,9 +18,11 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -41,7 +43,8 @@ var (
 // 3. Run data conversion (if schemaOnly is set to false)
 // 4. Generate report
 func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly, schemaOnly, skipForeignKeys bool, schemaSampleSize int64, sessionJSON string, ioHelper *utils.IOStreams, outputFilePrefix string, now time.Time) error {
-
+	// Cleanup hb tmp data directory in case residuals remain from prev runs.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	// Legacy mode is only supported for MySQL, PostgreSQL and DynamoDB
 	if driver != "" && utils.IsValidDriver(driver) && !utils.IsLegacyModeSupportedDriver(driver) {
 		return fmt.Errorf("legacy mode is not supported for drivers other than %s", strings.Join(utils.GetLegacyModeSupportedDrivers(), ", "))
@@ -97,7 +100,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(sourceProfile, targetProfile, ioHelper, client, conv, dataOnly)
+	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, ioHelper, client, conv, dataOnly)
 	if err != nil {
 		return fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 	}
@@ -109,5 +112,7 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, outputFilePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, outputFilePrefix+badDataFile, ioHelper.Out)
+	// Cleanup hb tmp data directory.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	return nil
 }
diff --git a/cmd/data.go b/cmd/data.go
index bd9883a..e88c06c 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -8,6 +8,7 @@ import (
 	"path"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -54,11 +55,13 @@ func (cmd *DataCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.sessionJSON, "session", "", "Specifies the file we restore session state from")
 	f.StringVar(&cmd.target, "target", "Spanner", "Specifies the target DB, defaults to Spanner (accepted values: `Spanner`)")
 	f.StringVar(&cmd.targetProfile, "target-profile", "", "Flag for specifying connection profile for target database e.g., \"dialect=postgresql\"")
-	flag.BoolVar(&cmd.skipForeignKeys, "skip-foreign-keys", false, "Skip creating foreign keys after data migration is complete (ddl statements for foreign keys can still be found in the downloaded schema.ddl.txt file and the same can be applied separately)")
+	f.BoolVar(&cmd.skipForeignKeys, "skip-foreign-keys", false, "Skip creating foreign keys after data migration is complete (ddl statements for foreign keys can still be found in the downloaded schema.ddl.txt file and the same can be applied separately)")
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
 }
 
 func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
+	// Cleanup hb tmp data directory in case residuals remain from prev runs.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	var err error
 	defer func() {
 		if err != nil {
@@ -136,7 +139,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		}
 	}
 
-	bw, err := conversion.DataConv(sourceProfile, targetProfile, &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true)
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
@@ -150,5 +153,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
+	// Cleanup hb tmp data directory.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	return subcommands.ExitSuccess
 }
diff --git a/cmd/schema.go b/cmd/schema.go
index 1d5f9cc..42d9a0c 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -8,6 +8,7 @@ import (
 	"path"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -55,6 +56,8 @@ func (cmd *SchemaCmd) SetFlags(f *flag.FlagSet) {
 }
 
 func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
+	// Cleanup hb tmp data directory in case residuals remain from prev runs.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	var err error
 	defer func() {
 		if err != nil {
@@ -106,5 +109,7 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 	conversion.WriteSchemaFile(conv, now, cmd.filePrefix+schemaFile, ioHelper.Out)
 	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
 	conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix+reportFile, ioHelper.Out)
+	// Cleanup hb tmp data directory.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	return subcommands.ExitSuccess
 }
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index b6bdd17..72b77eb 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -8,6 +8,7 @@ import (
 	"path"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -52,11 +53,13 @@ func (cmd *SchemaAndDataCmd) SetFlags(f *flag.FlagSet) {
 	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`)")
 	f.StringVar(&cmd.targetProfile, "target-profile", "", "Flag for specifying connection profile for target database e.g., \"dialect=postgresql\"")
-	flag.BoolVar(&cmd.skipForeignKeys, "skip-foreign-keys", false, "Skip creating foreign keys after data migration is complete (ddl statements for foreign keys can still be found in the downloaded schema.ddl.txt file and the same can be applied separately)")
+	f.BoolVar(&cmd.skipForeignKeys, "skip-foreign-keys", false, "Skip creating foreign keys after data migration is complete (ddl statements for foreign keys can still be found in the downloaded schema.ddl.txt file and the same can be applied separately)")
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
 }
 
 func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
+	// Cleanup hb tmp data directory in case residuals remain from prev runs.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	var err error
 	defer func() {
 		if err != nil {
@@ -134,7 +137,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(sourceProfile, targetProfile, &ioHelper, client, conv, true)
+	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true)
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
@@ -148,5 +151,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
+	// Cleanup hb tmp data directory.
+	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	return subcommands.ExitSuccess
 }
diff --git a/common/constants/constants.go b/common/constants/constants.go
index 83d5c61..1489818 100644
--- a/common/constants/constants.go
+++ b/common/constants/constants.go
@@ -26,6 +26,10 @@ const (
 	// CSV is the driver name when loading data using csv.
 	CSV string = "csv"
 
+	// ORACLE is the driver name for Oracle.
+	// This is an experimental driver; implementation in progress.
+	ORACLE string = "oracle"
+
 	// Target db for which schema is being generated.
 	TargetSpanner              string = "spanner"
 	TargetExperimentalPostgres string = "experimental_postgres"
@@ -33,4 +37,7 @@ const (
 	// Supported dialects for Cloud Spanner database.
 	DIALECT_POSTGRESQL string = "postgresql"
 	DIALECT_GOOGLESQL  string = "google_standard_sql"
+
+	// Temp directory name to write data which we cleanup at the end.
+	HB_TMP_DIR string = "harbourbridge_tmp_data"
 )
diff --git a/common/utils/utils.go b/common/utils/utils.go
index 4ba6456..1a1d193 100644
--- a/common/utils/utils.go
+++ b/common/utils/utils.go
@@ -13,6 +13,8 @@ import (
 	"net/url"
 	"os"
 	"os/exec"
+	"reflect"
+	"sort"
 	"strings"
 	"syscall"
 	"time"
@@ -22,6 +24,9 @@ import (
 	instance "cloud.google.com/go/spanner/admin/instance/apiv1"
 	"cloud.google.com/go/storage"
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/spanner"
 	"golang.org/x/crypto/ssh/terminal"
 	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
@@ -34,6 +39,12 @@ type IOStreams struct {
 	BytesRead           int64
 }
 
+// Harbourbridge accepts a manifest file in the form of a json which unmarshalls into the ManifestTables struct.
+type ManifestTable struct {
+	Table_name    string   `json:"table_name"`
+	File_patterns []string `json:"file_patterns"`
+}
+
 // NewIOStreams returns a new IOStreams struct such that input stream is set
 // to open file descriptor for dumpFile if driver is PGDUMP or MYSQLDUMP.
 // Input stream defaults to stdin. Output stream is always set to stdout.
@@ -51,7 +62,7 @@ func NewIOStreams(driver string, dumpFile string) IOStreams {
 		if u.Scheme == "gs" {
 			bucketName := u.Host
 			filePath := u.Path[1:] // removes "/" from beginning of path
-			f, err = downloadFromGCS(bucketName, filePath)
+			f, err = DownloadFromGCS(bucketName, filePath, "harbourbridge.gcs.data")
 		} else {
 			f, err = os.Open(dumpFile)
 		}
@@ -64,8 +75,8 @@ func NewIOStreams(driver string, dumpFile string) IOStreams {
 	return io
 }
 
-// downloadFromGCS returns the dump file that is downloaded from GCS
-func downloadFromGCS(bucketName string, filePath string) (*os.File, error) {
+// DownloadFromGCS returns the dump file that is downloaded from GCS.
+func DownloadFromGCS(bucketName, filePath, tmpFile string) (*os.File, error) {
 	ctx := context.Background()
 
 	client, err := storage.NewClient(ctx)
@@ -85,22 +96,23 @@ func downloadFromGCS(bucketName string, filePath string) (*os.File, error) {
 	defer rc.Close()
 	r := bufio.NewReader(rc)
 
-	tmpfile, err := ioutil.TempFile("", "harbourbridge.gcs.data")
+	tmpDir := os.TempDir() + constants.HB_TMP_DIR
+	os.MkdirAll(tmpDir, os.ModePerm)
+	tmpfile, err := os.Create(tmpDir + "/" + tmpFile)
 	if err != nil {
 		fmt.Printf("saveFile: unable to open temporary file to save dump file from GCS bucket %v", err)
 		log.Fatal(err)
 		return nil, err
 	}
-	syscall.Unlink(tmpfile.Name()) // File will be deleted when this process exits.
 
-	fmt.Printf("\nDownloading dump file from GCS bucket %s, path %s\n", bucketName, filePath)
+	fmt.Printf("\nDownloading file from GCS bucket %s, path %s\n", bucketName, filePath)
 	buffer := make([]byte, 1024)
 	for {
 		// read a chunk
 		n, err := r.Read(buffer[:cap(buffer)])
 
 		if err != nil && err != io.EOF {
-			fmt.Printf("readFile: unable to read entire dump file from bucket %s, file %s: %v", bucketName, filePath, err)
+			fmt.Printf("readFile: unable to read entire file from bucket %s, file %s: %v", bucketName, filePath, err)
 			log.Fatal(err)
 			return nil, err
 		}
@@ -118,6 +130,32 @@ func downloadFromGCS(bucketName string, filePath string) (*os.File, error) {
 	return tmpfile, nil
 }
 
+// PreloadGCSFiles downloads gcs files to tmp and updates the file paths in manifest with the local path.
+func PreloadGCSFiles(tables []ManifestTable) ([]ManifestTable, error) {
+	for i, table := range tables {
+		for j, filePath := range table.File_patterns {
+			u, err := url.Parse(filePath)
+			if err != nil {
+				return nil, fmt.Errorf("unable parse file path %s for table %s", filePath, table.Table_name)
+			}
+			if u.Scheme == "gs" {
+				bucketName := u.Host
+				filePath := u.Path[1:] // removes "/" from beginning of path
+				tmpFile := strings.ReplaceAll(filePath, "/", ".")
+				// Files get downloaded to tmp dir.
+				fileLoc := os.TempDir() + constants.HB_TMP_DIR + "/" + tmpFile
+				_, err = DownloadFromGCS(bucketName, filePath, tmpFile)
+				if err != nil {
+					return nil, fmt.Errorf("cannot download gcs file: %s for table %s", filePath, table.Table_name)
+				}
+				tables[i].File_patterns[j] = fileLoc
+				fmt.Printf("Downloaded file: %s\n", fileLoc)
+			}
+		}
+	}
+	return tables, nil
+}
+
 // GetProject returns the cloud project we should use for accessing Spanner.
 // Use environment variable GCLOUD_PROJECT if it is set.
 // Otherwise, use the default project returned from gcloud.
@@ -289,6 +327,15 @@ func ContainsAny(s string, l []string) bool {
 	return false
 }
 
+// CheckEqualSets checks if the set of values in a and b are equal.
+func CheckEqualSets(a, b []string) bool {
+	tmp_a := append(make([]string, len(a)), a...)
+	tmp_b := append(make([]string, len(b)), b...)
+	sort.Strings(tmp_a)
+	sort.Strings(tmp_b)
+	return reflect.DeepEqual(tmp_a, tmp_b)
+}
+
 func GetFileSize(f *os.File) (int64, error) {
 	info, err := f.Stat()
 	if err != nil {
@@ -415,3 +462,26 @@ func IsLegacyModeSupportedDriver(driver string) bool {
 func GetLegacyModeSupportedDrivers() []string {
 	return GetValidDrivers()[:5]
 }
+
+// ReadSpannerSchema fills conv by querying Spanner infoschema treating Spanner as both the source and dest.
+func ReadSpannerSchema(ctx context.Context, conv *internal.Conv, client *sp.Client) error {
+	infoSchema := spanner.InfoSchemaImpl{Client: client, Ctx: ctx, TargetDb: conv.TargetDb}
+	err := common.ProcessSchema(conv, infoSchema)
+	if err != nil {
+		return fmt.Errorf("error trying to read and convert spanner schema: %v", err)
+	}
+	parentTables, err := infoSchema.GetInterleaveTables()
+	if err != nil {
+		// We should ideally throw an error here as it could potentially cause a lot of failed writes.
+		// We raise an unexpected error for now to make it compatible with the integration tests.
+		// In the emulator, the interleave_type column in not supported hence the query fails.
+		conv.Unexpected(fmt.Sprintf("error trying to fetch interleave table info from schema: %v", err))
+	}
+	// Assign parents if any.
+	for table, parent := range parentTables {
+		spTable := conv.SpSchema[table]
+		spTable.Parent = parent
+		conv.SpSchema[table] = spTable
+	}
+	return nil
+}
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 9e2f3d9..f93e6a9 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -52,10 +52,11 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/sources/csv"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/dynamodb"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/mysql"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/oracle"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/postgres"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/sqlserver"
-	"github.com/cloudspannerecosystem/harbourbridge/spanner"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
 )
 
 var (
@@ -77,7 +78,7 @@ var (
 // When using source-profile, the sqlConnectionStr is constructed from the input params.
 func SchemaConv(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper *utils.IOStreams) (*internal.Conv, error) {
 	switch sourceProfile.Driver {
-	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER:
+	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB, constants.SQLSERVER, constants.ORACLE:
 		return schemaFromDatabase(sourceProfile, targetProfile)
 	case constants.PGDUMP, constants.MYSQLDUMP:
 		return schemaFromDump(sourceProfile.Driver, targetProfile.TargetDb, ioHelper)
@@ -92,8 +93,8 @@ func SchemaConv(sourceProfile profiles.SourceProfile, targetProfile profiles.Tar
 //  - Driver is DynamoDB or a dump file mode.
 //  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
 // When using source-profile, the sqlConnectionStr and schemaSampleSize are constructed from the input params.
-func DataConv(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper *utils.IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
-	config := spanner.BatchWriterConfig{
+func DataConv(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, ioHelper *utils.IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*writer.BatchWriter, error) {
+	config := writer.BatchWriterConfig{
 		BytesLimit: 100 * 1000 * 1000,
 		WriteLimit: 40,
 		RetryLimit: 1000,
@@ -108,7 +109,7 @@ func DataConv(sourceProfile profiles.SourceProfile, targetProfile profiles.Targe
 		}
 		return dataFromDump(sourceProfile.Driver, config, ioHelper, client, conv, dataOnly)
 	case constants.CSV:
-		return dataFromCSV(sourceProfile, targetProfile, config, conv, client)
+		return dataFromCSV(ctx, sourceProfile, targetProfile, config, conv, client)
 	default:
 		return nil, fmt.Errorf("data conversion for driver %s not supported", sourceProfile.Driver)
 	}
@@ -141,6 +142,8 @@ func connectionConfig(sourceProfile profiles.SourceProfile) (interface{}, error)
 		return getDynamoDBClientConfig()
 	case constants.SQLSERVER:
 		return profiles.GetSQLConnectionStr(sourceProfile), nil
+	case constants.ORACLE:
+		return profiles.GetSQLConnectionStr(sourceProfile), nil
 	default:
 		return "", fmt.Errorf("driver %s not supported", sourceProfile.Driver)
 	}
@@ -156,6 +159,11 @@ func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
 	case constants.SQLSERVER:
 		splts := strings.Split(sqlConnectionStr, "?database=")
 		return splts[len(splts)-1]
+	case constants.ORACLE:
+		// connection string formate : "oracle://user:password@104.108.154.85:1521/XE"
+		substr := sqlConnectionStr[9:]
+		dbName := strings.Split(substr, ":")[0]
+		return dbName
 	}
 	return ""
 }
@@ -170,7 +178,7 @@ func schemaFromDatabase(sourceProfile profiles.SourceProfile, targetProfile prof
 	return conv, common.ProcessSchema(conv, infoSchema)
 }
 
-func dataFromDatabase(sourceProfile profiles.SourceProfile, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
+func dataFromDatabase(sourceProfile profiles.SourceProfile, config writer.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*writer.BatchWriter, error) {
 	infoSchema, err := GetInfoSchema(sourceProfile)
 	if err != nil {
 		return nil, err
@@ -188,18 +196,18 @@ func dataFromDatabase(sourceProfile profiles.SourceProfile, config spanner.Batch
 		p.MaybeReport(atomic.LoadInt64(&rows))
 		return nil
 	}
-	writer := spanner.NewBatchWriter(config)
+	batchWriter := writer.NewBatchWriter(config)
 	conv.SetDataMode()
 	conv.SetDataSink(
 		func(table string, cols []string, vals []interface{}) {
-			writer.AddRow(table, cols, vals)
+			batchWriter.AddRow(table, cols, vals)
 		})
 	conv.DataFlush = func() {
-		writer.Flush()
+		batchWriter.Flush()
 	}
 	common.ProcessData(conv, infoSchema)
-	writer.Flush()
-	return writer, nil
+	batchWriter.Flush()
+	return batchWriter, nil
 }
 
 func getDynamoDBClientConfig() (*aws.Config, error) {
@@ -234,7 +242,7 @@ func schemaFromDump(driver string, targetDb string, ioHelper *utils.IOStreams) (
 	return conv, nil
 }
 
-func dataFromDump(driver string, config spanner.BatchWriterConfig, ioHelper *utils.IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
+func dataFromDump(driver string, config writer.BatchWriterConfig, ioHelper *utils.IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*writer.BatchWriter, error) {
 	// TODO: refactor of the way we handle getSeekable
 	// to avoid the code duplication here
 	if !dataOnly {
@@ -268,38 +276,48 @@ func dataFromDump(driver string, config spanner.BatchWriterConfig, ioHelper *uti
 		p.MaybeReport(atomic.LoadInt64(&rows))
 		return nil
 	}
-	writer := spanner.NewBatchWriter(config)
+	batchWriter := writer.NewBatchWriter(config)
 	conv.SetDataMode() // Process data in dump; schema is unchanged.
 	conv.SetDataSink(
 		func(table string, cols []string, vals []interface{}) {
-			writer.AddRow(table, cols, vals)
+			batchWriter.AddRow(table, cols, vals)
 		})
 	conv.DataFlush = func() {
-		writer.Flush()
+		batchWriter.Flush()
 	}
 	ProcessDump(driver, conv, r)
-	writer.Flush()
+	batchWriter.Flush()
 	p.Done()
 
-	return writer, nil
+	return batchWriter, nil
 }
 
-func dataFromCSV(sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, config spanner.BatchWriterConfig, conv *internal.Conv, client *sp.Client) (*spanner.BatchWriter, error) {
+func dataFromCSV(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client) (*writer.BatchWriter, error) {
 	if targetProfile.Conn.Sp.Dbname == "" {
 		return nil, fmt.Errorf("dbname is mandatory in target-profile for csv source")
 	}
+	conv.TargetDb = targetProfile.ToLegacyTargetDb()
 	delimiterStr := sourceProfile.Csv.Delimiter
 	if len(delimiterStr) != 1 {
 		return nil, fmt.Errorf("delimiter should only be a single character long, found '%s'", delimiterStr)
 	}
 	delimiter := rune(delimiterStr[0])
-	tables, err := csv.LoadManifest(conv, sourceProfile.Csv.Manifest)
+
+	err := utils.ReadSpannerSchema(ctx, conv, client)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error trying to read and convert spanner schema: %v", err)
+	}
+
+	tables, err := csv.GetCSVFiles(conv, sourceProfile)
+	if err != nil {
+		return nil, fmt.Errorf("error finding csv files: %v", err)
 	}
 
 	// Find the number of rows in each csv file for generating stats.
-	csv.SetRowStats(conv, tables, delimiter)
+	err = csv.SetRowStats(conv, tables, delimiter)
+	if err != nil {
+		return nil, err
+	}
 	totalRows := conv.Rows()
 	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
 	rows := int64(0)
@@ -312,19 +330,22 @@ func dataFromCSV(sourceProfile profiles.SourceProfile, targetProfile profiles.Ta
 		p.MaybeReport(atomic.LoadInt64(&rows))
 		return nil
 	}
-	writer := spanner.NewBatchWriter(config)
+	batchWriter := writer.NewBatchWriter(config)
 	conv.SetDataMode()
 	conv.SetDataSink(
 		func(table string, cols []string, vals []interface{}) {
-			writer.AddRow(table, cols, vals)
+			batchWriter.AddRow(table, cols, vals)
 		})
+	conv.DataFlush = func() {
+		batchWriter.Flush()
+	}
 	err = csv.ProcessCSV(conv, tables, sourceProfile.Csv.NullStr, delimiter)
 	if err != nil {
 		return nil, fmt.Errorf("can't process csv: %v", err)
 	}
-	writer.Flush()
+	batchWriter.Flush()
 	p.Done()
-	return writer, nil
+	return batchWriter, nil
 }
 
 // Report generates a report of schema and data conversion.
@@ -697,7 +718,7 @@ func ReadSessionFile(conv *internal.Conv, sessionJSON string) error {
 
 // WriteBadData prints summary stats about bad rows and writes detailed info
 // to file 'name'.
-func WriteBadData(bw *spanner.BatchWriter, conv *internal.Conv, banner, name string, out *os.File) {
+func WriteBadData(bw *writer.BatchWriter, conv *internal.Conv, banner, name string, out *os.File) {
 	badConversions := conv.BadRows()
 	badWrites := utils.SumMapValues(bw.DroppedRowsByTable())
 	if badConversions == 0 && badWrites == 0 {
@@ -787,6 +808,13 @@ func GetInfoSchema(sourceProfile profiles.SourceProfile) (common.InfoSchema, err
 			return nil, err
 		}
 		return sqlserver.InfoSchemaImpl{DbName: dbName, Db: db}, nil
+	case constants.ORACLE:
+		db, err := sql.Open(driver, connectionConfig.(string))
+		dbName := getDbNameFromSQLConnectionStr(driver, connectionConfig.(string))
+		if err != nil {
+			return nil, err
+		}
+		return oracle.InfoSchemaImpl{DbName: strings.ToUpper(dbName), Db: db}, nil
 	default:
 		return nil, fmt.Errorf("driver %s not supported", driver)
 	}
diff --git a/go.mod b/go.mod
index 1cff300..8cf9d99 100644
--- a/go.mod
+++ b/go.mod
@@ -22,6 +22,7 @@ require (
 	//github.com/pingcap/parser v3.0.12+incompatible
 	github.com/pingcap/parser v0.0.0-20200422082501-7329d80eaf2c
 	github.com/pingcap/tidb v1.1.0-beta.0.20200423105559-af376db3dc46
+	github.com/sijms/go-ora/v2 v2.2.17 // indirect
 	github.com/sirupsen/logrus v1.5.0 // indirect
 	github.com/stretchr/testify v1.6.1
 	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
diff --git a/go.sum b/go.sum
index 83aa8e0..e7104f2 100644
--- a/go.sum
+++ b/go.sum
@@ -403,6 +403,8 @@ github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749/go.mod h1:ZY1cvUeJ
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/shurcooL/vfsgen v0.0.0-20181020040650-a97a25d856ca h1:3fECS8atRjByijiI8yYiuwLwQ2ZxXobW7ua/8GRB3pI=
 github.com/shurcooL/vfsgen v0.0.0-20181020040650-a97a25d856ca/go.mod h1:TrYk7fJVaAttu97ZZKrO9UbRa8izdowaMIZcxYMbVaw=
+github.com/sijms/go-ora/v2 v2.2.17 h1:7w1lkgxorhhx/xG5fS/hWhLqBw9BrSFxTvx9oBj0Z0E=
+github.com/sijms/go-ora/v2 v2.2.17/go.mod h1:jzfAFD+4CXHE+LjGWFl6cPrtiIpQVxakI2gvrMF2w6Y=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.5.0 h1:1N5EYkVAPEywqZRJd7cwnRtCb6xJx7NH3T3WUTF980Q=
 github.com/sirupsen/logrus v1.5.0/go.mod h1:+F7Ogzej0PZc/94MaYx/nvG9jOFMD2osvC3s+Squfpo=
diff --git a/main.go b/main.go
index 9a2aa45..7563014 100644
--- a/main.go
+++ b/main.go
@@ -28,6 +28,7 @@ import (
 	_ "github.com/denisenkom/go-mssqldb"
 	_ "github.com/go-sql-driver/mysql"
 	_ "github.com/lib/pq"
+	_ "github.com/sijms/go-ora/v2"
 
 	"github.com/cloudspannerecosystem/harbourbridge/cmd"
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
diff --git a/profiles/common.go b/profiles/common.go
index e1ecb9c..673d96a 100644
--- a/profiles/common.go
+++ b/profiles/common.go
@@ -5,10 +5,12 @@ import (
 	"encoding/csv"
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
+	go_ora "github.com/sijms/go-ora/v2"
 )
 
 // Parses input string `s` as a map of key-value pairs. It's expected that the
@@ -96,6 +98,9 @@ func GetSQLConnectionStr(sourceProfile SourceProfile) string {
 		case SourceProfileConnectionTypeSqlServer:
 			connParams := sourceProfile.Conn.SqlServer
 			return getSQLSERVERConnectionStr(connParams.Host, connParams.Port, connParams.User, connParams.Pwd, connParams.Db)
+		case SourceProfileConnectionTypeOracle:
+			connParams := sourceProfile.Conn.Oracle
+			return getORACLEConnectionStr(connParams.Host, connParams.Port, connParams.User, connParams.Pwd, connParams.Db)
 		}
 	}
 	return sqlConnectionStr
@@ -156,3 +161,8 @@ func GetSchemaSampleSize(sourceProfile SourceProfile) int64 {
 	}
 	return schemaSampleSize
 }
+
+func getORACLEConnectionStr(server, port, user, password, dbname string) string {
+	portNumber, _ := strconv.Atoi(port)
+	return go_ora.BuildUrl(server, portNumber, dbname, user, password, nil)
+}
diff --git a/profiles/source_profile.go b/profiles/source_profile.go
index cdb5582..e335e14 100644
--- a/profiles/source_profile.go
+++ b/profiles/source_profile.go
@@ -48,6 +48,7 @@ const (
 	SourceProfileConnectionTypePostgreSQL
 	SourceProfileConnectionTypeDynamoDB
 	SourceProfileConnectionTypeSqlServer
+	SourceProfileConnectionTypeOracle
 )
 
 type SourceProfileConnectionMySQL struct {
@@ -267,12 +268,52 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 	return dydb, nil
 }
 
+type SourceProfileConnectionOracle struct {
+	Host string
+	Port string
+	User string
+	Db   string
+	Pwd  string
+}
+
+func NewSourceProfileConnectionOracle(params map[string]string) (SourceProfileConnectionOracle, error) {
+	ss := SourceProfileConnectionOracle{}
+	host, hostOk := params["host"]
+	user, userOk := params["user"]
+	db, dbOk := params["db_name"]
+	port, _ := params["port"]
+	pwd, _ := params["password"]
+
+	if hostOk && userOk && dbOk {
+		// All connection params provided through source-profile. Port and password handled later.
+		ss.Host, ss.User, ss.Db, ss.Port, ss.Pwd = host, user, db, port, pwd
+		// Throw error if the input entered is empty.
+		if ss.Host == "" || ss.User == "" || ss.Db == "" {
+			return ss, fmt.Errorf("found empty string for host/user/db_name. Please specify host, port, user and db_name in the source-profile")
+		}
+	} else {
+		// Partial params provided through source-profile. Ask user to provide all through the source-profile.
+		return ss, fmt.Errorf("please specify host, port, user and db_name in the source-profile")
+	}
+
+	if ss.Port == "" {
+		// Set default port for oracle, which rarely changes.
+		ss.Port = "1521"
+	}
+	if ss.Pwd == "" {
+		ss.Pwd = utils.GetPassword()
+	}
+
+	return ss, nil
+}
+
 type SourceProfileConnection struct {
 	Ty        SourceProfileConnectionType
 	Mysql     SourceProfileConnectionMySQL
 	Pg        SourceProfileConnectionPostgreSQL
 	Dydb      SourceProfileConnectionDynamoDB
 	SqlServer SourceProfileConnectionSqlServer
+	Oracle    SourceProfileConnectionOracle
 }
 
 func NewSourceProfileConnection(source string, params map[string]string) (SourceProfileConnection, error) {
@@ -312,7 +353,14 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 				return conn, err
 			}
 		}
-
+	case "oracle":
+		{
+			conn.Ty = SourceProfileConnectionTypeOracle
+			conn.Oracle, err = NewSourceProfileConnectionOracle(params)
+			if err != nil {
+				return conn, err
+			}
+		}
 	default:
 		return conn, fmt.Errorf("please specify a valid source database using -source flag, received source = %v", source)
 	}
@@ -392,6 +440,8 @@ func (src SourceProfile) ToLegacyDriver(source string) (string, error) {
 				return constants.DYNAMODB, nil
 			case "sqlserver", "mssql":
 				return constants.SQLSERVER, nil
+			case "oracle":
+				return constants.ORACLE, nil
 			default:
 				return "", fmt.Errorf("please specify a valid source database using -source flag, received source = %v", source)
 			}
@@ -431,12 +481,7 @@ func NewSourceProfile(s string, source string) (SourceProfile, error) {
 		return SourceProfile{}, fmt.Errorf("could not parse source-profile, error = %v", err)
 	}
 	if strings.ToLower(source) == constants.CSV {
-		if _, ok := params["manifest"]; ok {
-			profile := NewSourceProfileCsv(params)
-			return SourceProfile{Ty: SourceProfileTypeCsv, Csv: profile}, nil
-		} else {
-			return SourceProfile{}, fmt.Errorf("csv source requires a manifest file, please specify manifest file in the source profile e.g., -source-profile=\"manifest=file_path\"")
-		}
+		return SourceProfile{Ty: SourceProfileTypeCsv, Csv: NewSourceProfileCsv(params)}, nil
 	}
 
 	if _, ok := params["file"]; ok || filePipedToStdin() {
diff --git a/sources/csv/data.go b/sources/csv/data.go
index 9b1fad5..131396d 100644
--- a/sources/csv/data.go
+++ b/sources/csv/data.go
@@ -23,34 +23,52 @@ import (
 	"math/big"
 	"os"
 	"strconv"
+	"strings"
 	"time"
 
 	"cloud.google.com/go/civil"
+	"cloud.google.com/go/spanner"
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
-	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
-type Column struct {
-	Column_name string `json:"column_name"`
-	Type_name   string `json:"type_name"`
-}
+// GetCSVFiles finds the appropriate files paths and downloads gcs files in any.
+func GetCSVFiles(conv *internal.Conv, sourceProfile profiles.SourceProfile) (tables []utils.ManifestTable, err error) {
+	// If manifest file not provided, we assume the csvs exist in the same directory
+	// in table_name.csv format.
+	if sourceProfile.Csv.Manifest == "" {
+		fmt.Println("Manifest file not provided, checking for files named `[table_name].csv` in current working directory...")
+		for t := range conv.SpSchema {
+			tables = append(tables, utils.ManifestTable{Table_name: t, File_patterns: []string{fmt.Sprintf("%s.csv", t)}})
+		}
+	} else {
+		fmt.Println("Manifest file provided, reading csv file paths...")
+		// Read paths provided in manifest.
+		tables, err = loadManifest(conv, sourceProfile.Csv.Manifest)
+		if err != nil {
+			return nil, err
+		}
+	}
 
-// Harbourbridge accepts a manifest file in the form of a json which unmarshalls into the Table struct.
-type Table struct {
-	Table_name    string   `json:"table_name"`
-	File_patterns []string `json:"file_patterns"`
-	Columns       []Column `json:"columns"`
+	// Download gcs files if any.
+	tables, err = utils.PreloadGCSFiles(tables)
+	if err != nil {
+		return nil, fmt.Errorf("gcs file download error: %v", err)
+	}
+	return tables, nil
 }
 
-// LoadManifest reads the manifest file and unmarshalls it into a list of Table struct.
+// loadManifest reads the manifest file and unmarshalls it into a list of Table struct.
 // It also performs certain checks on the manifest.
-func LoadManifest(conv *internal.Conv, manifestFile string) ([]Table, error) {
+func loadManifest(conv *internal.Conv, manifestFile string) ([]utils.ManifestTable, error) {
 	manifest, err := ioutil.ReadFile(manifestFile)
 	if err != nil {
 		return nil, fmt.Errorf("can't read manifest file due to: %v", err)
 	}
-	tables := []Table{}
+	tables := []utils.ManifestTable{}
 	err = json.Unmarshal(manifest, &tables)
 	if err != nil {
 		return nil, fmt.Errorf("unable to unmarshall json due to: %v", err)
@@ -64,70 +82,83 @@ func LoadManifest(conv *internal.Conv, manifestFile string) ([]Table, error) {
 
 // VerifyManifest performs certain prechecks on the structure of the manifest while populating the conv with
 // the ddl types. Also checks on valid file paths and empty CSVs are handled as conv.Unexpected errors later during processing.
-func VerifyManifest(conv *internal.Conv, tables []Table) error {
+func VerifyManifest(conv *internal.Conv, tables []utils.ManifestTable) error {
 	if len(tables) == 0 {
 		return fmt.Errorf("no tables found")
 	}
+	missing := []string{}
+	for name := range conv.SrcSchema {
+		found := false
+		for _, table := range tables {
+			if name == table.Table_name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			missing = append(missing, name)
+		}
+	}
+	if len(missing) > 0 {
+		fmt.Printf("WARNING: did not find manifest entries for tables [ %s ], ignoring and proceeding...\n", strings.Join(missing, ", "))
+		conv.Unexpected(fmt.Sprintf("did not find manifest entries for tables [ %s ]", strings.Join(missing, ", ")))
+	}
 	for i, table := range tables {
 		name := table.Table_name
 		if name == "" {
 			return fmt.Errorf("table number %d (0-indexed) does not have a name", i)
 		}
+		if _, ok := conv.SrcSchema[name]; !ok {
+			return fmt.Errorf("table %s provided in manifest does not exist in spanner", name)
+		}
 		if len(table.File_patterns) == 0 {
 			return fmt.Errorf("no file path provided for table %s", name)
 		}
-		cols := table.Columns
-		if len(cols) == 0 {
-			return fmt.Errorf("`columns` field for table %s is empty", name)
-		}
-		// Populating just the table names in the conv for SrcSchema and SpSchema
-		// so the report for row stats is generated.
-		conv.SrcSchema[table.Table_name] = schema.Table{Name: table.Table_name}
-
-		// The map colDefs stores the mapping from column names to their final types.
-		colDefs := make(map[string]ddl.ColumnDef)
-		for j, col := range cols {
-			if col.Column_name == "" || col.Type_name == "" {
-				return fmt.Errorf("please provide column_name and type_name in `columns` field at position %d (0-indexed)", j)
-			}
-			ty, err := ToSpannerType(col.Type_name)
-			if err != nil {
-				return fmt.Errorf("can't map to spanner type: %v. Please use the data types as in your spanner database", err)
-			}
-			colDefs[col.Column_name] = ddl.ColumnDef{Name: col.Column_name, T: ty}
-		}
-		conv.SpSchema[table.Table_name] = ddl.CreateTable{Name: table.Table_name, ColDefs: colDefs}
 	}
 	return nil
 }
 
 // SetRowStats calculates the number of rows per table.
-func SetRowStats(conv *internal.Conv, tables []Table, delimiter rune) {
+func SetRowStats(conv *internal.Conv, tables []utils.ManifestTable, delimiter rune) error {
 	for _, table := range tables {
 		for _, filePath := range table.File_patterns {
 			csvFile, err := os.Open(filePath)
 			if err != nil {
-				fmt.Printf("can't read csv file: %s due to: %v\n", filePath, err)
+				return fmt.Errorf("can't read csv file: %s due to: %v", filePath, err)
 			}
 			r := csvReader.NewReader(csvFile)
 			r.Comma = delimiter
-			count, err := getCSVRowCount(r)
+			count, err := getCSVDataRowCount(r, conv.SpSchema[table.Table_name].ColNames)
 			if err != nil {
-				conv.Unexpected(fmt.Sprintf("Couldn't get number of rows for table %s", table.Table_name))
-				continue
+				return fmt.Errorf("error reading file %s for table %s: %v", filePath, table.Table_name, err)
 			}
 			if count == 0 {
 				conv.Unexpected(fmt.Sprintf("error processing table %s: file %s is empty.", table.Table_name, filePath))
 				continue
 			}
-			conv.Stats.Rows[table.Table_name] += count - 1
+			conv.Stats.Rows[table.Table_name] += count
 		}
 	}
+	return nil
 }
 
-// getCSVRowCount returns the number of rows in the CSV file.
-func getCSVRowCount(r *csvReader.Reader) (int64, error) {
+// getCSVDataRowCount returns the number of data rows in the CSV file. This excludes the headers if present.
+func getCSVDataRowCount(r *csvReader.Reader, colNames []string) (int64, error) {
 	count := int64(0)
+	srcCols, err := r.Read()
+	if err == io.EOF {
+		return count, nil
+	}
+	if err != nil {
+		return count, fmt.Errorf("can't read csv headers for col names due to: %v", err)
+	}
+	if len(srcCols) != len(colNames) {
+		return 0, fmt.Errorf("found %d columns in csv, expected %d as per Spanner schema", len(srcCols), len(colNames))
+	}
+	// If the row read was not a header, increase count.
+	if !utils.CheckEqualSets(srcCols, colNames) {
+		count += 1
+	}
 	for {
 		_, err := r.Read()
 		if err == io.EOF {
@@ -143,8 +174,18 @@ func getCSVRowCount(r *csvReader.Reader) (int64, error) {
 
 // ProcessCSV writes data across the tables provided in the manifest file. Each table's data can be provided
 // across multiple CSV files hence, the manifest accepts a list of file paths in the input.
-func ProcessCSV(conv *internal.Conv, tables []Table, nullStr string, delimiter rune) error {
+func ProcessCSV(conv *internal.Conv, tables []utils.ManifestTable, nullStr string, delimiter rune) error {
+	orderedTableNames := ddl.OrderTables(conv.SpSchema)
+	nameToFiles := map[string][]string{}
 	for _, table := range tables {
+		nameToFiles[table.Table_name] = table.File_patterns
+	}
+	orderedTables := []utils.ManifestTable{}
+	for _, name := range orderedTableNames {
+		orderedTables = append(orderedTables, utils.ManifestTable{name, nameToFiles[name]})
+	}
+
+	for _, table := range orderedTables {
 		for _, filePath := range table.File_patterns {
 			csvFile, err := os.Open(filePath)
 			if err != nil {
@@ -153,26 +194,38 @@ func ProcessCSV(conv *internal.Conv, tables []Table, nullStr string, delimiter r
 			r := csvReader.NewReader(csvFile)
 			r.Comma = delimiter
 
-			// First row is expected to be the column headers.
+			// Default column order is same as in Spanner schema.
+			colNames := conv.SpSchema[table.Table_name].ColNames
 			srcCols, err := r.Read()
 			if err == io.EOF {
 				conv.Unexpected(fmt.Sprintf("error processing table %s: file %s is empty.", table.Table_name, filePath))
 				continue
 			}
 			if err != nil {
-				return fmt.Errorf("can't read csv headers for col names due to: %v", err)
+				return fmt.Errorf("can't read row for %s due to: %v", filePath, err)
 			}
+			// If first row is some permutation of Spanner schema columns, we assume the first row is headers.
+			if utils.CheckEqualSets(srcCols, colNames) {
+				colNames = srcCols
+			} else {
+				// Write the first row since it was not a column header.
+				processDataRow(conv, nullStr, table.Table_name, colNames, srcCols)
+			}
+
 			for {
 				values, err := r.Read()
 				if err == io.EOF {
 					break
 				}
 				if err != nil {
-					return fmt.Errorf(fmt.Sprintf("can't read row  names due to: %v", err))
+					return fmt.Errorf("can't read row for %s due to: %v", filePath, err)
 				}
-				processDataRow(conv, nullStr, table.Table_name, srcCols, values)
+				processDataRow(conv, nullStr, table.Table_name, colNames, values)
 			}
 		}
+		if conv.DataFlush != nil {
+			conv.DataFlush()
+		}
 	}
 	return nil
 }
@@ -200,7 +253,14 @@ func convertData(conv *internal.Conv, nullStr, tableName string, srcCols []strin
 			continue
 		}
 		colName := srcCols[i]
-		x, err := convScalar(colDefs[colName].T, val)
+		spColDef := colDefs[colName]
+		var x interface{}
+		var err error
+		if spColDef.T.IsArray {
+			x, err = convArray(spColDef.T, val)
+		} else {
+			x, err = convScalar(conv, spColDef.T, val)
+		}
 		if err != nil {
 			return nil, nil, err
 		}
@@ -210,7 +270,169 @@ func convertData(conv *internal.Conv, nullStr, tableName string, srcCols []strin
 	return cvtCols, v, nil
 }
 
-func convScalar(spannerType ddl.Type, val string) (interface{}, error) {
+func convArray(spannerType ddl.Type, val string) (interface{}, error) {
+	val = strings.TrimSpace(val)
+	// Handle empty array. Note that we use an empty NullString array
+	// for all Spanner array types since this will be converted to the
+	// appropriate type by the Spanner client.
+	if val == "{}" || val == "[]" {
+		return []spanner.NullString{}, nil
+	}
+	braces := val[:1] + val[len(val)-1:]
+	if braces != "{}" && braces != "[]" {
+		return []interface{}{}, fmt.Errorf("unrecognized data format for array: expected {v1, v2, ...} or [v1, v2, ...]")
+	}
+	a := strings.Split(val[1:len(val)-1], ",")
+
+	// The Spanner client for go does not accept []interface{} for arrays.
+	// Instead it only accepts slices of a specific type e.g. []int64, []string.
+	// Hence we have to do the following case analysis.
+	switch spannerType.Name {
+	case ddl.Bool:
+		var r []spanner.NullBool
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullBool{Valid: false})
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return []spanner.NullBool{}, err
+			}
+			b, err := convBool(s)
+			if err != nil {
+				return []spanner.NullBool{}, err
+			}
+			r = append(r, spanner.NullBool{Bool: b, Valid: true})
+		}
+		return r, nil
+	case ddl.Bytes:
+		var r [][]byte
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, nil)
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return [][]byte{}, err
+			}
+			b, err := convBytes(s)
+			if err != nil {
+				return [][]byte{}, err
+			}
+			r = append(r, b)
+		}
+		return r, nil
+	case ddl.Date:
+		var r []spanner.NullDate
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullDate{Valid: false})
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return []spanner.NullDate{}, err
+			}
+			date, err := convDate(s)
+			if err != nil {
+				return []spanner.NullDate{}, err
+			}
+			r = append(r, spanner.NullDate{Date: date, Valid: true})
+		}
+		return r, nil
+	case ddl.Float64:
+		var r []spanner.NullFloat64
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullFloat64{Valid: false})
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return []spanner.NullFloat64{}, err
+			}
+			f, err := convFloat64(s)
+			if err != nil {
+				return []spanner.NullFloat64{}, err
+			}
+			r = append(r, spanner.NullFloat64{Float64: f, Valid: true})
+		}
+		return r, nil
+	case ddl.Numeric:
+		var r []spanner.NullNumeric
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullNumeric{Valid: false})
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return []spanner.NullNumeric{}, err
+			}
+			n, err := convNumeric(s)
+			if err != nil {
+				return []spanner.NullNumeric{}, err
+			}
+			r = append(r, spanner.NullNumeric{Numeric: n, Valid: true})
+		}
+		return r, nil
+	case ddl.Int64:
+		var r []spanner.NullInt64
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullInt64{Valid: false})
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return []spanner.NullInt64{}, err
+			}
+			i, err := convInt64(s)
+			if err != nil {
+				return r, err
+			}
+			r = append(r, spanner.NullInt64{Int64: i, Valid: true})
+		}
+		return r, nil
+	case ddl.String:
+		var r []spanner.NullString
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullString{Valid: false})
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return []spanner.NullString{}, err
+			}
+			r = append(r, spanner.NullString{StringVal: s, Valid: true})
+		}
+		return r, nil
+	case ddl.Timestamp:
+		var r []spanner.NullTime
+		for _, s := range a {
+			if s == "NULL" {
+				r = append(r, spanner.NullTime{Valid: false})
+				continue
+			}
+			s, err := processQuote(s)
+			if err != nil {
+				return []spanner.NullTime{}, err
+			}
+			t, err := convTimestamp(s)
+			if err != nil {
+				return []spanner.NullTime{}, err
+			}
+			r = append(r, spanner.NullTime{Time: t, Valid: true})
+		}
+		return r, nil
+	}
+	return []interface{}{}, fmt.Errorf("array type conversion not implemented for type []%v", spannerType.Name)
+}
+
+func convScalar(conv *internal.Conv, spannerType ddl.Type, val string) (interface{}, error) {
 	switch spannerType.Name {
 	case ddl.Bool:
 		return convBool(val)
@@ -223,6 +445,9 @@ func convScalar(spannerType ddl.Type, val string) (interface{}, error) {
 	case ddl.Int64:
 		return convInt64(val)
 	case ddl.Numeric:
+		if conv.TargetDb == constants.TargetExperimentalPostgres {
+			return spanner.PGNumeric{Numeric: val, Valid: true}, nil
+		}
 		return convNumeric(val)
 	case ddl.String:
 		return val, nil
@@ -275,12 +500,12 @@ 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(val string) (interface{}, error) {
+func convNumeric(val string) (big.Rat, error) {
 	r := new(big.Rat)
 	if _, ok := r.SetString(val); !ok {
-		return "", fmt.Errorf("can't convert %q to big.Rat", val)
+		return big.Rat{}, fmt.Errorf("can't convert %q to big.Rat", val)
 	}
-	return r, nil
+	return *r, nil
 }
 
 func convTimestamp(val string) (t time.Time, err error) {
@@ -290,3 +515,10 @@ func convTimestamp(val string) (t time.Time, err error) {
 	}
 	return t, err
 }
+
+func processQuote(s string) (string, error) {
+	if len(s) >= 2 && s[0] == '"' && s[len(s)-1] == '"' {
+		return strconv.Unquote(s)
+	}
+	return s, nil
+}
diff --git a/sources/oracle/infoschema.go b/sources/oracle/infoschema.go
new file mode 100644
index 0000000..07d5722
--- /dev/null
+++ b/sources/oracle/infoschema.go
@@ -0,0 +1,296 @@
+package oracle
+
+import (
+	"database/sql"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+type InfoSchemaImpl struct {
+	DbName string
+	Db     *sql.DB
+}
+
+// GetToDdl function below 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
+}
+
+// GetRowsFromTable returns a sql Rows object for a table.
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
+	panic("Not Implemented")
+}
+
+// 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 {
+	panic("Not Implemented")
+}
+
+func ProcessDataRow(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, values []string) {
+	panic("Not Implemented")
+}
+
+// GetRowCount with number of rows in each table.
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
+	panic("Not Implemented")
+}
+
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
+	q := fmt.Sprintf("SELECT table_name FROM all_tables WHERE owner = '%s'", isi.DbName)
+	rows, err := isi.Db.Query(q)
+	if err != nil {
+		return nil, fmt.Errorf("couldn't get tables: %w", err)
+	}
+	defer rows.Close()
+	var tableName string
+	var tables []common.SchemaAndName
+	for rows.Next() {
+		rows.Scan(&tableName)
+		tables = append(tables, common.SchemaAndName{Schema: isi.DbName, Name: tableName})
+	}
+	return tables, nil
+}
+
+// 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 := fmt.Sprintf(`
+					SELECT 
+						column_name, 
+						data_type, 
+						nullable, 
+						data_default, 
+						data_length, 
+						data_precision, 
+						data_scale 
+					FROM all_tab_columns 
+					WHERE owner = '%s' AND table_name = '%s'
+					`, table.Schema, table.Name)
+	cols, err := isi.Db.Query(q)
+	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 string
+	var isNullable string
+	var colDefault sql.NullString
+	var charMaxLen, numericPrecision, numericScale sql.NullInt64
+	for cols.Next() {
+		err := cols.Scan(&colName, &dataType, &isNullable, &colDefault, &charMaxLen, &numericPrecision, &numericScale)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		ignored := schema.Ignored{}
+		for _, c := range constraints[colName] {
+			// Type of constraint definition in oracle C (check constraint on a table)
+			// P (primary key), U (unique key) ,R (referential integrity), V (with check option, on a view)
+			// O (with read only, on a view).
+			// We've already filtered out PRIMARY KEY.
+			switch c {
+			case "C":
+				ignored.Check = true
+			case "R", "P", "U":
+				// Nothing to do here -- these are handled elsewhere.
+			}
+		}
+		ignored.Default = colDefault.Valid
+		c := schema.Column{
+			Name:    colName,
+			Type:    toType(dataType, charMaxLen, numericPrecision, numericScale),
+			NotNull: strings.ToUpper(isNullable) == "N",
+			Ignored: ignored,
+		}
+		colDefs[colName] = c
+		colNames = append(colNames, colName)
+	}
+	return colDefs, colNames, nil
+}
+
+// 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, table common.SchemaAndName) ([]string, map[string][]string, error) {
+	q := fmt.Sprintf(`
+					SELECT 
+						k.column_name,
+						t.constraint_type
+	   				FROM all_constraints t
+       				INNER JOIN all_cons_columns k
+       				ON (k.constraint_name = t.constraint_name) 
+					WHERE t.owner = '%s' AND k.table_name = '%s'
+					`, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+	var primaryKeys []string
+	var col, constraint string
+	m := make(map[string][]string)
+	for rows.Next() {
+		err := rows.Scan(&col, &constraint)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		if col == "" || constraint == "" {
+			conv.Unexpected("Got empty col or constraint")
+			continue
+		}
+		// P (primary key) constraint in oracle
+		switch constraint {
+		case "P":
+			primaryKeys = append(primaryKeys, col)
+		default:
+			m[col] = append(m[col], constraint)
+		}
+	}
+	return primaryKeys, m, nil
+}
+
+// GetForeignKeys return list all the foreign keys constraints.
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+	q := fmt.Sprintf(`
+						SELECT 
+							B.table_name AS ref_table, 
+							A.column_name AS col_name,
+							B.column_name AS ref_col_name,
+							A.constraint_name AS name
+						FROM all_cons_columns A 
+						JOIN all_constraints C ON A.owner = C.owner AND A.constraint_name = C.constraint_name
+						JOIN all_cons_columns B ON B.owner = C.owner AND B.constraint_name = C.r_constraint_name
+						WHERE A.table_name='%s' AND A.owner='%s'
+					`, table.Name, isi.DbName)
+	rows, err := isi.Db.Query(q)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var col, refCol, refTable, fKeyName string
+	fKeys := make(map[string]common.FkConstraint)
+	var keyNames []string
+
+	for rows.Next() {
+		err := rows.Scan(&refTable, &col, &refCol, &fKeyName)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		if _, found := fKeys[fKeyName]; found {
+			fk := fKeys[fKeyName]
+			fk.Cols = append(fk.Cols, col)
+			fk.Refcols = append(fk.Refcols, refCol)
+			fKeys[fKeyName] = fk
+			continue
+		}
+		fKeys[fKeyName] = common.FkConstraint{Name: fKeyName, Table: refTable, Refcols: []string{refCol}, Cols: []string{col}}
+		keyNames = append(keyNames, fKeyName)
+	}
+	sort.Strings(keyNames)
+	for _, k := range keyNames {
+		foreignKeys = append(foreignKeys,
+			schema.ForeignKey{
+				Name:         fKeys[k].Name,
+				Columns:      fKeys[k].Cols,
+				ReferTable:   fKeys[k].Table,
+				ReferColumns: fKeys[k].Refcols})
+	}
+	return foreignKeys, nil
+}
+
+// GetIndexes return a list of all indexes for the specified table.
+// Oracle db support several types of index:
+// 1. Normal indexes. (By default, Oracle Database creates B-tree indexes.)
+// 2.Bitmap indexes
+// 3.Partitioned indexes
+// 4. Function-based indexes
+// 5.Domain indexes,
+// we are only considering normal index as of now.
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
+	q := fmt.Sprintf(`
+					SELECT 
+						IC.index_name,
+						IC.column_name,
+						IC.column_position, 
+						IC.descend,
+						I.uniqueness, 
+						IE.column_expression, 
+						I.index_type 
+                	FROM  all_ind_columns IC 
+					LEFT JOIN all_ind_expressions IE ON IC.index_name = IE.index_name AND IC.column_position=IE.column_position
+                	LEFT JOIN all_indexes I ON IC.index_name = I.index_name
+                	WHERE IC.index_owner='%s' AND IC.table_name='%s'
+            		ORDER BY IC.index_name, IC.column_position
+				`, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var name, column, sequence, Unique, indexType string
+	var collation, colexpression sql.NullString
+	indexMap := make(map[string]schema.Index)
+	var indexNames []string
+	ignoredIndex := make(map[string]bool)
+	var indexes []schema.Index
+	for rows.Next() {
+		if err := rows.Scan(&name, &column, &sequence, &collation, &Unique, &colexpression, &indexType); err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		// ingnore all index except normal
+		// UPPER("EMAIL") check for the function call with "(",")"
+		if indexType != "NORMAL" && strings.Contains(colexpression.String, "(") && strings.Contains(colexpression.String, ")") {
+			ignoredIndex[name] = true
+		}
+
+		//INDEX1_LAST	SYS_NC00009$	1	DESC	NONUNIQUE	"LAST_NAME"	FUNCTION-BASED NORMAL
+		// DESC column make index functional index but as special case we included that
+		// and update column name with column expression
+		if colexpression.Valid && !strings.Contains(colexpression.String, "(") && !strings.Contains(colexpression.String, ")") {
+			column = colexpression.String[1 : len(colexpression.String)-1]
+		}
+
+		if _, found := indexMap[name]; !found {
+			indexNames = append(indexNames, name)
+			indexMap[name] = schema.Index{Name: name, Unique: (Unique == "UNIQUE")}
+		}
+		index := indexMap[name]
+		index.Keys = append(index.Keys, schema.Key{Column: column, Desc: (collation.Valid && collation.String == "DESC")})
+		indexMap[name] = index
+	}
+	for _, k := range indexNames {
+		// only add noraml index
+		if _, found := ignoredIndex[k]; !found {
+			indexes = append(indexes, indexMap[k])
+		}
+	}
+	return indexes, nil
+}
+
+func toType(dataType string, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
+	switch {
+	case dataType == "NUMBER" && numericPrecision.Valid && numericScale.Valid && numericScale.Int64 != 0:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64, numericScale.Int64}}
+	case dataType == "NUMBER" && numericPrecision.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64}}
+	// Oracle get column query return data length for the Number type.
+	case dataType != "NUMBER" && charLen.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{charLen.Int64}}
+	default:
+		return schema.Type{Name: dataType}
+	}
+}
diff --git a/sources/oracle/toddl.go b/sources/oracle/toddl.go
new file mode 100644
index 0000000..e97ae80
--- /dev/null
+++ b/sources/oracle/toddl.go
@@ -0,0 +1,113 @@
+// Copyright 2021 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package oracle handles schema and data migrations from oracle.
+package oracle
+
+import (
+	"regexp"
+
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+var (
+	timestampReg = regexp.MustCompile(`TIMESTAMP`)
+	intervalReg  = regexp.MustCompile(`INTERVAL`)
+)
+
+// ToDdlImpl oracle specific implementation for ToDdl.
+type ToDdlImpl struct {
+}
+
+// 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.TargetExperimentalPostgres {
+		ty = overrideExperimentalType(columnType, ty)
+	}
+	return ty, issues
+}
+
+func toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Type, []internal.SchemaIssue) {
+	// Oracle returns some datatype with the precision,
+	// So will get TIMESTAMP as TIMESTAMP(6),TIMESTAMP(6) WITH TIME ZONE,TIMESTAMP(6) WITH LOCAL TIME ZONE.
+	// To match this case timestampReg Regex defined.
+	if timestampReg.MatchString(id) {
+		return ddl.Type{Name: ddl.Timestamp}, nil
+	}
+
+	// Matching cases like INTERVAL YEAR(2) TO MONTH, INTERVAL DAY(2) TO SECOND(6),etc.
+	if intervalReg.MatchString(id) {
+		if len(mods) > 0 {
+			return ddl.Type{Name: ddl.String, Len: 30}, nil
+		}
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	}
+
+	switch id {
+	case "NUMBER":
+		// If no scale is avalible then map it to int64, and numeric elsewhere.
+		if len(mods) == 1 && mods[0] >= 1 && mods[0] < 19 {
+			return ddl.Type{Name: ddl.Int64}, nil
+		} else {
+			return ddl.Type{Name: ddl.Numeric}, nil
+		}
+	case "BFILE", "BLOB":
+		return ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, nil
+	case "CHAR":
+		return ddl.Type{Name: ddl.String, Len: mods[0]}, nil
+	case "CLOB":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "DATE":
+		return ddl.Type{Name: ddl.Date}, nil
+	case "BINARY_DOUBLE", "BINARY_FLOAT", "FLOAT":
+		return ddl.Type{Name: ddl.Float64}, nil
+	case "LONG":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "RAW", "LONG RAW":
+		return ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}, nil
+	case "NCHAR", "NVARCHAR2", "VARCHAR", "VARCHAR2":
+		if len(mods) > 0 {
+			return ddl.Type{Name: ddl.String, Len: mods[0]}, nil
+		}
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "NCLOB":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "ROWID":
+		return ddl.Type{Name: ddl.String, Len: 10}, nil
+	case "UROWID":
+		if len(mods) > 0 {
+			return ddl.Type{Name: ddl.String, Len: mods[0]}, nil
+		}
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	case "XMLTYPE":
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, nil
+	default:
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.NoGoodType}
+	}
+}
+
+// Override the types to map to experimental postgres types.
+func overrideExperimentalType(columnType schema.Type, originalType ddl.Type) ddl.Type {
+	if columnType.Name == "DATE" {
+		return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}
+	}
+	return originalType
+}
diff --git a/sources/spanner/infoschema.go b/sources/spanner/infoschema.go
new file mode 100644
index 0000000..5cc701c
--- /dev/null
+++ b/sources/spanner/infoschema.go
@@ -0,0 +1,343 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package spanner
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"sort"
+	"strconv"
+	"strings"
+
+	"cloud.google.com/go/spanner"
+	_ "github.com/lib/pq" // we will use database/sql package instead of using this package directly
+	"google.golang.org/api/iterator"
+
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+// InfoSchemaImpl postgres specific implementation for InfoSchema.
+type InfoSchemaImpl struct {
+	Client   *spanner.Client
+	Ctx      context.Context
+	TargetDb string
+}
+
+// GetToDdl function below implement the common.InfoSchema interface.
+func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
+	return ToDdlImpl{}
+}
+
+// We leave the 3 functions below empty to be able to pass this as an infoSchema interface. We don't need these for now.
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	return nil
+}
+
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
+	return 0, nil
+}
+
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
+	return nil, nil
+}
+
+// GetTableName returns table name.
+func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
+	if isi.TargetDb == constants.TargetExperimentalPostgres {
+		if schema == "public" { // Drop public prefix for pg spanner.
+			return tableName
+		}
+	} else {
+		if schema == "" {
+			return tableName
+		}
+	}
+	return fmt.Sprintf("%s.%s", schema, tableName)
+}
+
+// GetTables return list of tables in the selected database.
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
+	q := `SELECT table_schema, table_name FROM information_schema.tables 
+	WHERE table_type = 'BASE TABLE' AND table_schema = ''`
+	if isi.TargetDb == constants.TargetExperimentalPostgres {
+		q = `SELECT table_schema, table_name FROM information_schema.tables 
+	WHERE table_type = 'BASE TABLE' AND table_schema = 'public'`
+	}
+	stmt := spanner.Statement{SQL: q}
+	iter := isi.Client.Single().Query(isi.Ctx, stmt)
+	defer iter.Stop()
+
+	var tableSchema, tableName string
+	var tables []common.SchemaAndName
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, fmt.Errorf("couldn't get tables: %w", err)
+		}
+		err = row.Columns(&tableSchema, &tableName)
+		if err != nil {
+			return nil, err
+		}
+		tables = append(tables, common.SchemaAndName{Schema: tableSchema, Name: tableName})
+	}
+	return tables, nil
+}
+
+// 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 column_name, spanner_type, is_nullable 
+			FROM information_schema.columns
+			WHERE table_schema = '' AND table_name = @p1
+			ORDER BY ordinal_position;`
+	if isi.TargetDb == constants.TargetExperimentalPostgres {
+		q = `SELECT column_name, spanner_type, is_nullable 
+			FROM information_schema.columns
+			WHERE table_schema = 'public' AND table_name = $1
+			ORDER BY ordinal_position;`
+	}
+	stmt := spanner.Statement{
+		SQL: q,
+		Params: map[string]interface{}{
+			"p1": table.Name,
+		},
+	}
+	iter := isi.Client.Single().Query(isi.Ctx, stmt)
+	defer iter.Stop()
+
+	colDefs := make(map[string]schema.Column)
+	var colNames []string
+	var colName, spannerType, isNullable string
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, nil, fmt.Errorf("couldn't get column info for table %s: %s", table.Name, err)
+		}
+		err = row.Columns(&colName, &spannerType, &isNullable)
+		if err != nil {
+			return nil, nil, fmt.Errorf("cannot read row for table %s while reading columns: %s", table.Name, err)
+		}
+		ignored := schema.Ignored{}
+		for _, c := range constraints[colName] {
+			switch c {
+			case "CHECK":
+				ignored.Check = true
+			case "FOREIGN KEY", "PRIMARY KEY", "UNIQUE":
+				// Nothing to do here -- these are handled elsewhere.
+			}
+		}
+		c := schema.Column{
+			Name:    colName,
+			Type:    toType(spannerType),
+			NotNull: common.ToNotNull(conv, isNullable),
+		}
+		colDefs[colName] = c
+		colNames = append(colNames, colName)
+	}
+	return colDefs, colNames, nil
+}
+
+// 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, 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 = '' AND k.table_name = @p1 ORDER BY k.ordinal_position;`
+	if isi.TargetDb == constants.TargetExperimentalPostgres {
+		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 = 'public' AND k.table_name = $1 ORDER BY k.ordinal_position;`
+	}
+	stmt := spanner.Statement{
+		SQL: q,
+		Params: map[string]interface{}{
+			"p1": table.Name,
+		},
+	}
+	iter := isi.Client.Single().Query(isi.Ctx, stmt)
+	defer iter.Stop()
+
+	var primaryKeys []string
+	var col, constraint string
+	m := make(map[string][]string)
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, nil, fmt.Errorf("couldn't get row while reading constraints: %w", err)
+		}
+		err = row.Columns(&col, &constraint)
+		if err != nil {
+			return nil, nil, err
+		}
+		if col == "" || constraint == "" {
+			conv.Unexpected("Got empty col or constraint")
+			continue
+		}
+		switch constraint {
+		case "PRIMARY KEY":
+			primaryKeys = append(primaryKeys, col)
+		default:
+			m[col] = append(m[col], constraint)
+		}
+	}
+	return primaryKeys, m, nil
+}
+
+// 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  k.constraint_name, k.column_name, t.table_name, c.column_name 
+			FROM information_schema.key_column_usage AS k 
+			JOIN information_schema.constraint_column_usage AS c ON k.constraint_name = c.constraint_name
+			JOIN information_schema.table_constraints AS t ON k.constraint_name = t.constraint_name 
+			WHERE t.constraint_type='FOREIGN KEY' AND t.table_schema = '' AND t.table_name = @p1
+			ORDER BY k.constraint_name, k.ordinal_position;`
+	if isi.TargetDb == constants.TargetExperimentalPostgres {
+		q = `SELECT  k.constraint_name, k.column_name, t.table_name, c.column_name 
+				FROM information_schema.key_column_usage AS k 
+				JOIN information_schema.constraint_column_usage AS c ON k.constraint_name = c.constraint_name
+				JOIN information_schema.table_constraints AS t ON k.constraint_name = t.constraint_name 
+				WHERE t.constraint_type='FOREIGN KEY' AND t.table_schema = 'public' AND t.table_name = $1
+				ORDER BY k.constraint_name, k.ordinal_position;`
+	}
+	stmt := spanner.Statement{
+		SQL: q,
+		Params: map[string]interface{}{
+			"p1": table.Name,
+		},
+	}
+	iter := isi.Client.Single().Query(isi.Ctx, stmt)
+	defer iter.Stop()
+
+	var col, refCol, fKeyName, refTable string
+	fKeys := make(map[string]common.FkConstraint)
+	var keyNames []string
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, fmt.Errorf("couldn't get row while fetching foreign keys: %w", err)
+		}
+		err = row.Columns(&fKeyName, &col, &refTable, &refCol)
+		if err != nil {
+			return nil, err
+		}
+		if _, found := fKeys[fKeyName]; found {
+			fk := fKeys[fKeyName]
+			fk.Cols = append(fk.Cols, col)
+			fk.Refcols = append(fk.Refcols, refCol)
+			fKeys[fKeyName] = fk
+			continue
+		}
+		fKeys[fKeyName] = common.FkConstraint{Name: fKeyName, Table: isi.GetTableName(table.Schema, refTable), Refcols: []string{refCol}, Cols: []string{col}}
+		keyNames = append(keyNames, fKeyName)
+	}
+	sort.Strings(keyNames)
+	for _, k := range keyNames {
+		// The query returns a crypted result for multi-col FKs. Currently for a FK from (a,b,c) -> (x,y,z),
+		// the returned rows like (a,x), (a,y), (a,z), (b,x), (b,y), (b,z), (c,x), (c,y), (c,z).
+		// Need to reduce it to (a,x), (b,y), (c,z). The logic below does that.
+		n := int(math.Sqrt(float64(len(fKeys[k].Cols))))
+		cols, refcols := []string{}, []string{}
+		for i := 0; i < n; i++ {
+			cols = append(cols, fKeys[k].Cols[i*n])
+			refcols = append(refcols, fKeys[k].Refcols[i])
+		}
+		foreignKeys = append(foreignKeys,
+			schema.ForeignKey{
+				Name:         fKeys[k].Name,
+				Columns:      cols,
+				ReferTable:   fKeys[k].Table,
+				ReferColumns: refcols})
+	}
+	return foreignKeys, nil
+}
+
+// Skipped since we dont have a use case for storing indexes yet.
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
+	return []schema.Index{}, nil
+}
+
+func (isi InfoSchemaImpl) GetInterleaveTables() (map[string]string, error) {
+	q := `SELECT table_name, parent_table_name FROM information_schema.tables 
+	WHERE interleave_type = 'IN PARENT' AND table_type = 'BASE TABLE' AND table_schema = ''`
+	if isi.TargetDb == constants.TargetExperimentalPostgres {
+		q = `SELECT table_name, parent_table_name FROM information_schema.tables 
+		WHERE interleave_type = 'IN PARENT' AND table_type = 'BASE TABLE' AND table_schema = 'public'`
+	}
+	stmt := spanner.Statement{SQL: q}
+	iter := isi.Client.Single().Query(isi.Ctx, stmt)
+	defer iter.Stop()
+
+	var tableName, parentTable string
+	parentTables := map[string]string{}
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, fmt.Errorf("couldn't read row while fetching interleaved tables: %w", err)
+		}
+		err = row.Columns(&tableName, &parentTable)
+		if err != nil {
+			return nil, err
+		}
+		parentTables[tableName] = parentTable
+	}
+	return parentTables, nil
+}
+
+func toType(dataType string) schema.Type {
+	switch {
+	case strings.Contains(dataType, "ARRAY"):
+		typeLenStr := dataType[(strings.Index(dataType, "<") + 1):(len(dataType) - 1)]
+		schemaType := toType(typeLenStr)
+		schemaType.ArrayBounds = []int64{-1}
+		return schemaType
+	case strings.Contains(dataType, "("):
+		idx := strings.Index(dataType, "(")
+		typeLenStr := dataType[(idx + 1):(len(dataType) - 1)]
+		var typeLen int64
+		if typeLenStr == "MAX" {
+			typeLen = ddl.MaxLength
+		} else {
+			typeLen, _ = strconv.ParseInt(typeLenStr, 10, 64)
+		}
+		return schema.Type{Name: dataType[:idx], Mods: []int64{typeLen}}
+	default:
+		return schema.Type{Name: dataType}
+	}
+}
diff --git a/sources/spanner/toddl.go b/sources/spanner/toddl.go
new file mode 100644
index 0000000..0f5ca2c
--- /dev/null
+++ b/sources/spanner/toddl.go
@@ -0,0 +1,79 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package spanner
+
+import (
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+type ToDdlImpl struct {
+}
+
+// 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.TargetExperimentalPostgres {
+		ty, issues = overrideExperimentalType(columnType, ty, issues)
+	} else {
+		ty.IsArray = len(columnType.ArrayBounds) == 1
+	}
+	return ty, issues
+}
+
+// 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 toSpannerTypeInternal(conv *internal.Conv, id string, mods []int64) (ddl.Type, []internal.SchemaIssue) {
+	switch id {
+	case "BOOL":
+		return ddl.Type{Name: ddl.Bool}, nil
+	case "BYTES":
+		return ddl.Type{Name: ddl.Bytes, Len: mods[0]}, nil
+	case "DATE":
+		return ddl.Type{Name: ddl.Date}, nil
+	case "FLOAT64":
+		return ddl.Type{Name: ddl.Float64}, nil
+	case "INT64":
+		return ddl.Type{Name: ddl.Int64}, nil
+	case "JSON":
+		return ddl.Type{Name: ddl.JSON}, nil
+	case "NUMERIC":
+		return ddl.Type{Name: ddl.Numeric}, nil
+	case "STRING":
+		return ddl.Type{Name: ddl.String, Len: mods[0]}, nil
+	case "TIMESTAMP":
+		return ddl.Type{Name: ddl.Timestamp}, nil
+	}
+	return ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, []internal.SchemaIssue{internal.NoGoodType}
+}
+
+// Override the types to map to experimental postgres types.
+func overrideExperimentalType(columnType schema.Type, originalType ddl.Type, issues []internal.SchemaIssue) (ddl.Type, []internal.SchemaIssue) {
+	switch columnType.Name {
+	case "PG.NUMERIC":
+		return ddl.Type{Name: ddl.Numeric}, nil
+	case "PG.JSONB":
+		return ddl.Type{Name: ddl.JSON}, nil
+	}
+	return originalType, issues
+}
diff --git a/spanner/batchwriter.go b/spanner/writer/batchwriter.go
similarity index 99%
rename from spanner/batchwriter.go
rename to spanner/writer/batchwriter.go
index b27030b..c82a78b 100644
--- a/spanner/batchwriter.go
+++ b/spanner/writer/batchwriter.go
@@ -15,7 +15,7 @@
 // Package spanner provides high-level abstractions for working with
 // Cloud Spanner that are not available from the core Cloud Spanner
 // libraries.
-package spanner
+package writer
 
 import (
 	"fmt"


Test patch:
diff --git a/.github/workflows/integration-tests-against-emulator.yaml b/.github/workflows/integration-tests-against-emulator.yaml
index 3df45d3..b139b25 100644
--- a/.github/workflows/integration-tests-against-emulator.yaml
+++ b/.github/workflows/integration-tests-against-emulator.yaml
@@ -118,7 +118,7 @@ jobs:
       # run tests
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.13'
+          go-version: '1.16'
       - run: go version
       - run: go test -v ./...
         env:
diff --git a/sources/csv/data_test.go b/sources/csv/data_test.go
index 39ea565..0b2fc9b 100644
--- a/sources/csv/data_test.go
+++ b/sources/csv/data_test.go
@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"cloud.google.com/go/civil"
+	"cloud.google.com/go/spanner"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
@@ -29,31 +31,15 @@ const (
 	SINGERS_2_CSV string = SINGERS_TABLE + "_2.csv"
 )
 
-func getManifestTables() []Table {
-	return []Table{
+func getManifestTables() []utils.ManifestTable {
+	return []utils.ManifestTable{
 		{
 			Table_name:    ALL_TYPES_TABLE,
 			File_patterns: []string{ALL_TYPES_CSV},
-			Columns: []Column{
-				{Column_name: "bool_col", Type_name: "BOOL"},
-				{Column_name: "byte_col", Type_name: "BYTES"},
-				{Column_name: "date_col", Type_name: "DATE"},
-				{Column_name: "float_col", Type_name: "FLOAT64"},
-				{Column_name: "int_col", Type_name: "INT64"},
-				{Column_name: "numeric_col", Type_name: "NUMERIC"},
-				{Column_name: "string_col", Type_name: "STRING"},
-				{Column_name: "timestamp_col", Type_name: "TIMESTAMP"},
-				{Column_name: "json_col", Type_name: "JSON"},
-			},
 		},
 		{
 			Table_name:    SINGERS_TABLE,
 			File_patterns: []string{SINGERS_1_CSV, SINGERS_2_CSV},
-			Columns: []Column{
-				{Column_name: "SingerId", Type_name: "INT64"},
-				{Column_name: "FirstName", Type_name: "STRING"},
-				{Column_name: "LastName", Type_name: "STRING"},
-			},
 		},
 	}
 }
@@ -103,34 +89,32 @@ func cleanupCSVs() {
 }
 
 func TestSetRowStats(t *testing.T) {
-	conv := internal.MakeConv()
+	conv := buildConv(getCreateTable())
 	writeCSVs(t)
 	defer cleanupCSVs()
 	SetRowStats(conv, getManifestTables(), ',')
 	assert.Equal(t, map[string]int64{ALL_TYPES_TABLE: 1, SINGERS_TABLE: 2}, conv.Stats.Rows)
 }
 
-func TestProcessDataRow(t *testing.T) {
-	conv := internal.MakeConv()
+func TestProcessCSV(t *testing.T) {
+	writeCSVs(t)
+	defer cleanupCSVs()
+	tables := getManifestTables()
+
+	conv := buildConv(getCreateTable())
 	var rows []spannerData
 	conv.SetDataMode()
 	conv.SetDataSink(
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-
-	writeCSVs(t)
-	defer cleanupCSVs()
-	tables := getManifestTables()
-	VerifyManifest(conv, tables)
 	err := ProcessCSV(conv, tables, "", ',')
-	fmt.Println(err)
 	assert.Nil(t, err)
 	assert.Equal(t, []spannerData{
 		{
 			table: ALL_TYPES_TABLE,
 			cols:  []string{"bool_col", "byte_col", "date_col", "float_col", "int_col", "numeric_col", "string_col", "timestamp_col", "json_col"},
-			vals:  []interface{}{true, []uint8{0x74, 0x65, 0x73, 0x74}, getDate("2019-10-29"), 15.13, int64(100), big.NewRat(3994, 100), "Helloworld", getTime(t, "2019-10-29T05:30:00Z"), "{\"key1\": \"value1\", \"key2\": \"value2\"}"},
+			vals:  []interface{}{true, []uint8{0x74, 0x65, 0x73, 0x74}, getDate("2019-10-29"), 15.13, int64(100), *big.NewRat(3994, 100), "Helloworld", getTime(t, "2019-10-29T05:30:00Z"), "{\"key1\": \"value1\", \"key2\": \"value2\"}"},
 		},
 		{table: SINGERS_TABLE, cols: []string{"SingerId", "FirstName", "LastName"}, vals: []interface{}{int64(1), "fn1", "ln1"}},
 		{table: SINGERS_TABLE, cols: []string{"SingerId", "FirstName", "LastName"}, vals: []interface{}{int64(2), "fn2", "ln2"}},
@@ -150,18 +134,21 @@ func TestConvertData(t *testing.T) {
 		{"date", ddl.Type{Name: ddl.Date}, "2019-10-29", getDate("2019-10-29")},
 		{"float64", ddl.Type{Name: ddl.Float64}, "42.6", float64(42.6)},
 		{"int64", ddl.Type{Name: ddl.Int64}, "42", int64(42)},
-		{"numeric", ddl.Type{Name: ddl.Numeric}, "42.6", big.NewRat(426, 10)},
+		{"numeric", ddl.Type{Name: ddl.Numeric}, "42.6", *big.NewRat(426, 10)},
 		{"string", ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, "eh", "eh"},
 		{"timestamp", ddl.Type{Name: ddl.Timestamp}, "2019-10-29 05:30:00", getTime(t, "2019-10-29T05:30:00Z")},
 		{"json", ddl.Type{Name: ddl.JSON}, "{\"key1\": \"value1\"}", "{\"key1\": \"value1\"}"},
+		{"int_array", ddl.Type{Name: ddl.Int64, IsArray: true}, "{1,2,NULL}", []spanner.NullInt64{{Int64: int64(1), Valid: true}, {Int64: int64(2), Valid: true}, {Valid: false}}},
+		{"string_array", ddl.Type{Name: ddl.String, IsArray: true}, "[ab,cd]", []spanner.NullString{{StringVal: "ab", Valid: true}, {StringVal: "cd", Valid: true}}},
+		{"float_array", ddl.Type{Name: ddl.Float64, IsArray: true}, "{1.3,2.5}", []spanner.NullFloat64{{Float64: float64(1.3), Valid: true}, {Float64: float64(2.5), Valid: true}}},
+		{"numeric_array", ddl.Type{Name: ddl.Numeric, IsArray: true}, "[1.7]", []spanner.NullNumeric{{Numeric: *big.NewRat(17, 10), Valid: true}}},
 	}
 	tableName := "testtable"
 	for _, tc := range singleColTests {
 		col := "a"
-		conv := buildConv(
-			ddl.CreateTable{
-				Name:    tableName,
-				ColDefs: map[string]ddl.ColumnDef{col: ddl.ColumnDef{Name: col, T: tc.ty}}})
+		conv := buildConv([]ddl.CreateTable{{
+			Name:    tableName,
+			ColDefs: map[string]ddl.ColumnDef{col: ddl.ColumnDef{Name: col, T: tc.ty}}}})
 		_, av, err := convertData(conv, "", tableName, []string{col}, []string{tc.in})
 		// NULL scenario.
 		if tc.ev == nil {
@@ -200,15 +187,46 @@ func TestConvertData(t *testing.T) {
 		},
 	}
 	for _, tc := range errorTests {
-		conv := buildConv(spTable)
+		conv := buildConv([]ddl.CreateTable{spTable})
 		_, _, err := convertData(conv, "", tableName, cols, tc.vals)
 		assert.NotNil(t, err, tc.name)
 	}
 }
 
-func buildConv(spTable ddl.CreateTable) *internal.Conv {
+func getCreateTable() []ddl.CreateTable {
+	return []ddl.CreateTable{
+		{
+			Name:     ALL_TYPES_TABLE,
+			ColNames: []string{"bool_col", "byte_col", "date_col", "float_col", "int_col", "numeric_col", "string_col", "timestamp_col", "json_col"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"bool_col":      {Name: "bool_col", T: ddl.Type{Name: ddl.Bool}},
+				"byte_col":      {Name: "byte_col", T: ddl.Type{Name: ddl.Bytes}},
+				"date_col":      {Name: "date_col", T: ddl.Type{Name: ddl.Date}},
+				"float_col":     {Name: "float_col", T: ddl.Type{Name: ddl.Float64}},
+				"int_col":       {Name: "int_col", T: ddl.Type{Name: ddl.Int64}},
+				"numeric_col":   {Name: "numeric_col", T: ddl.Type{Name: ddl.Numeric}},
+				"string_col":    {Name: "string_col", T: ddl.Type{Name: ddl.String}},
+				"timestamp_col": {Name: "timestamp_col", T: ddl.Type{Name: ddl.Timestamp}},
+				"json_col":      {Name: "json_col", T: ddl.Type{Name: ddl.JSON}},
+			},
+		},
+		{
+			Name:     SINGERS_TABLE,
+			ColNames: []string{"SingerId", "FirstName", "LastName"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"SingerId":  {Name: "SingerId", T: ddl.Type{Name: ddl.Int64}},
+				"FirstName": {Name: "FirstName", T: ddl.Type{Name: ddl.String}},
+				"LastName":  {Name: "LastName", T: ddl.Type{Name: ddl.String}},
+			},
+		},
+	}
+}
+
+func buildConv(spTables []ddl.CreateTable) *internal.Conv {
 	conv := internal.MakeConv()
-	conv.SpSchema[spTable.Name] = spTable
+	for _, spTable := range spTables {
+		conv.SpSchema[spTable.Name] = spTable
+	}
 	return conv
 }
 
diff --git a/sources/oracle/infoschema_test.go b/sources/oracle/infoschema_test.go
new file mode 100644
index 0000000..de93d18
--- /dev/null
+++ b/sources/oracle/infoschema_test.go
@@ -0,0 +1,183 @@
+// Copyright 2021 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package oracle
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
+)
+
+type mockSpec struct {
+	query string
+	args  []driver.Value   // Query args.
+	cols  []string         // Columns names for returned rows.
+	rows  [][]driver.Value // Set of rows returned.
+}
+
+func TestProcessSchemaMYSQL(t *testing.T) {
+	ms := []mockSpec{
+		{
+			query: "SELECT table_name FROM all_tables (.+)",
+			args:  []driver.Value{},
+			cols:  []string{"table_name"},
+			rows: [][]driver.Value{
+				{"USER"},
+				{"TEST"},
+			},
+		},
+		// USER table
+		{
+			query: `SELECT (.+) FROM all_constraints (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"column_name", "contraint_type"},
+			rows: [][]driver.Value{
+				{"USER_ID", "P"},
+				{"REF", "F"},
+			},
+		},
+		{
+			query: `SELECT (.+) all_cons_columns A JOIN all_constraints C ON (.+) JOIN all_cons_columns B (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"ref_table", "column_name", "ref_column_name", "name"},
+			rows: [][]driver.Value{
+				{"TEST", "REF", "ID", "fk_test"},
+			},
+		},
+		{
+			query: `SELECT (.+) LEFT JOIN all_ind_expressions IE (.+) LEFT JOIN all_indexes I (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"name", "column_name", "column_position", "descend", "uniqueness", "column_expression", "index_type"},
+			rows: [][]driver.Value{
+				{"INDEX1_LAST", "SYS_NC00009$", 1, "DESC", "NONUNIQUE", "\"NAME\"", "FUNCTION-BASED NORMAL"},
+				{"INDEX_CONTACT_TEST", "SYS_NC00008$", 1, "ASC", "UNIQUE", "UPPER(\"USER_ID\")", "FUNCTION-BASED NORMAL"},
+				{"INDEX_CONTACT_TEST", "REF", 2, "ASC", "UNIQUE", nil, "FUNCTION-BASED NORMAL"},
+				{"INDEX_CONTACT_TEST", "NAME", 3, "ASC", "UNIQUE", nil, "FUNCTION-BASED NORMAL"},
+				{"INDEX_TEST_2", "SYS_NC00007$", 1, "DESC", "NONUNIQUE", "\"NAME\"", "FUNCTION-BASED NORMAL"},
+				{"INDEX_TEST_2", "SYS_NC00009$", 2, "DESC", "NONUNIQUE", "\"USER_ID\"", "FUNCTION-BASED NORMAL"},
+			},
+		},
+		{
+			query: "SELECT (.+) FROM all_tab_columns (.+)",
+			args:  []driver.Value{},
+			cols:  []string{"column_name", "data_type", "nullable", "data_default", "data_length", "data_precision", "data_scale"},
+			rows: [][]driver.Value{
+				{"USER_ID", "VARCHAR2", "N", nil, nil, nil, nil},
+				{"NAME", "VARCHAR2", "N", nil, nil, nil, nil},
+				{"REF", "NUMBER", "Y", nil, nil, nil, nil}},
+		},
+
+		// test table
+		{
+			query: `SELECT (.+) FROM all_constraints (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"column_name", "contraint_type"},
+			rows: [][]driver.Value{
+				{"ID", "P"}},
+		},
+		{
+			query: `SELECT (.+) all_cons_columns A JOIN all_constraints C ON (.+) JOIN all_cons_columns B (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"ref_table", "column_name", "ref_column_name", "name"},
+			rows:  [][]driver.Value{},
+		},
+		{
+			query: `SELECT (.+) LEFT JOIN all_ind_expressions IE (.+) LEFT JOIN all_indexes I (.+)`,
+			args:  []driver.Value{},
+			cols:  []string{"name", "column_name", "column_position", "descend", "uniqueness", "column_expression", "index_type"},
+			rows:  [][]driver.Value{},
+		},
+		{
+			query: "SELECT (.+) FROM all_tab_columns (.+)",
+			args:  []driver.Value{},
+			cols:  []string{"column_name", "data_type", "nullable", "data_default", "data_length", "data_precision", "data_scale"},
+			rows: [][]driver.Value{
+				{"ID", "NUMBER", "N", nil, nil, nil, nil}},
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := internal.MakeConv()
+	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
+	assert.Nil(t, err)
+	expectedSchema := map[string]ddl.CreateTable{
+		"USER": {
+			Name:     "USER",
+			ColNames: []string{"USER_ID", "NAME", "REF"},
+			ColDefs:  map[string]ddl.ColumnDef{"USER_ID": {Name: "USER_ID", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength, IsArray: false}, NotNull: true}, "NAME": {Name: "NAME", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength, IsArray: false}, NotNull: true}, "REF": {Name: "REF", T: ddl.Type{Name: ddl.Numeric}}},
+			Pks:      []ddl.IndexKey{{Col: "USER_ID"}},
+			Fks:      []ddl.Foreignkey{{Name: "fk_test", Columns: []string{"REF"}, ReferTable: "TEST", ReferColumns: []string{"ID"}}},
+			Indexes: []ddl.CreateIndex{{
+				Name:   "INDEX1_LAST",
+				Table:  "USER",
+				Unique: false,
+				Keys:   []ddl.IndexKey{{Col: "NAME", Desc: true}},
+			}, {
+				Name:   "INDEX_TEST_2",
+				Table:  "USER",
+				Unique: false,
+				Keys:   []ddl.IndexKey{{Col: "NAME", Desc: true}, {Col: "USER_ID", Desc: true}},
+			}},
+		},
+		"TEST": {
+			Name:     "TEST",
+			ColNames: []string{"ID"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"ID": {Name: "ID", T: ddl.Type{Name: ddl.Numeric}, NotNull: true}},
+			Pks: []ddl.IndexKey{{Col: "ID"}},
+		},
+	}
+	assert.Equal(t, expectedSchema, stripSchemaComments(conv.SpSchema))
+	assert.Equal(t, len(conv.Issues["USER"]), 0)
+	assert.Equal(t, len(conv.Issues["TEST"]), 0)
+	assert.Equal(t, int64(0), conv.Unexpecteds())
+}
+
+// stripSchemaComments returns a schema with all comments removed.
+// We mostly ignore schema comments in testing since schema comments
+// are often changed and are not a core part of conversion functionality.
+func stripSchemaComments(spSchema map[string]ddl.CreateTable) map[string]ddl.CreateTable {
+	for t, ct := range spSchema {
+		for c, cd := range ct.ColDefs {
+			cd.Comment = ""
+			ct.ColDefs[c] = cd
+		}
+		ct.Comment = ""
+		spSchema[t] = ct
+	}
+	return spSchema
+}
+
+func mkMockDB(t *testing.T, ms []mockSpec) *sql.DB {
+	db, mock, err := sqlmock.New()
+	assert.Nil(t, err)
+	for _, m := range ms {
+		rows := sqlmock.NewRows(m.cols)
+		for _, r := range m.rows {
+			rows.AddRow(r...)
+		}
+		if len(m.args) > 0 {
+			mock.ExpectQuery(m.query).WithArgs(m.args...).WillReturnRows(rows)
+		} else {
+			mock.ExpectQuery(m.query).WillReturnRows(rows)
+		}
+	}
+	return db
+}
diff --git a/sources/oracle/toddl_test.go b/sources/oracle/toddl_test.go
new file mode 100644
index 0000000..67df390
--- /dev/null
+++ b/sources/oracle/toddl_test.go
@@ -0,0 +1,109 @@
+// Copyright 2021 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package oracle
+
+import (
+	"testing"
+
+	"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"
+)
+
+func TestToSpannerType(t *testing.T) {
+	conv := internal.MakeConv()
+	conv.SetSchemaMode()
+	name := "test"
+	srcSchema := schema.Table{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h"},
+		ColDefs: map[string]schema.Column{
+			"a": {Name: "a", Type: schema.Type{Name: "NUMBER"}},
+			"b": {Name: "b", Type: schema.Type{Name: "FLOAT"}},
+			"c": {Name: "c", Type: schema.Type{Name: "BFILE"}},
+			"d": {Name: "d", Type: schema.Type{Name: "VARCHAR2", Mods: []int64{20}}},
+			"e": {Name: "e", Type: schema.Type{Name: "DATE"}},
+			"f": {Name: "f", Type: schema.Type{Name: "TIMESTAMP"}},
+			"g": {Name: "g", Type: schema.Type{Name: "LONG"}},
+			"h": {Name: "h", Type: schema.Type{Name: "NUMBER", Mods: []int64{13}}},
+		},
+		PrimaryKeys: []schema.Key{{Column: "a"}},
+		ForeignKeys: []schema.ForeignKey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_fake", Columns: []string{"x"}, ReferTable: "ref_table", ReferColumns: []string{"x"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aRef"}}},
+		Indexes: []schema.Index{{Name: "index1", Unique: true, Keys: []schema.Key{{Column: "a", Desc: false}, {Column: "d", Desc: true}}},
+			{Name: "index_with_0_key", Unique: true, Keys: []schema.Key{{Column: "m", Desc: false}, {Column: "y", Desc: true}}},
+		},
+	}
+	conv.SrcSchema[name] = srcSchema
+	conv.SpSchema["ref_table"] = ddl.CreateTable{
+		Name:     "ref_table",
+		ColNames: []string{"dref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"dref": {Name: "dref", T: ddl.Type{Name: ddl.String, Len: int64(6)}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "dref"}},
+	}
+	conv.SpSchema["ref_table2"] = ddl.CreateTable{
+		Name:     "ref_table2",
+		ColNames: []string{"aref", "b", "c"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"aref": {Name: "aref", T: ddl.Type{Name: ddl.Numeric}},
+			"b":    {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c":    {Name: "c", T: ddl.Type{Name: ddl.Bool}},
+		},
+		Pks: []ddl.IndexKey{{Col: "aref"}},
+	}
+	assert.Nil(t, common.SchemaToSpannerDDL(conv, ToDdlImpl{}))
+	actual := conv.SpSchema[name]
+	dropComments(&actual) // Don't test comment.
+	expected := ddl.CreateTable{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f", "g", "h"},
+		ColDefs: map[string]ddl.ColumnDef{
+			"a": {Name: "a", T: ddl.Type{Name: ddl.Numeric}},
+			"b": {Name: "b", T: ddl.Type{Name: ddl.Float64}},
+			"c": {Name: "c", T: ddl.Type{Name: ddl.Bytes, Len: ddl.MaxLength}},
+			"d": {Name: "d", T: ddl.Type{Name: ddl.String, Len: int64(20)}},
+			"e": {Name: "e", T: ddl.Type{Name: ddl.Date}},
+			"f": {Name: "f", T: ddl.Type{Name: ddl.Timestamp}},
+			"g": {Name: "g", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+			"h": {Name: "h", T: ddl.Type{Name: ddl.Int64}},
+		},
+		Pks: []ddl.IndexKey{{Col: "a"}},
+		Fks: []ddl.Foreignkey{{Name: "fk_test", Columns: []string{"d"}, ReferTable: "ref_table", ReferColumns: []string{"dref"}},
+			{Name: "fk_test2", Columns: []string{"a"}, ReferTable: "ref_table2", ReferColumns: []string{"aref"}}},
+		Indexes: []ddl.CreateIndex{{Name: "index1", Table: name, Unique: true, Keys: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "d", Desc: true}}},
+			{Name: "index_with_0_key", Table: name, Unique: true, Keys: nil}},
+	}
+	assert.Equal(t, expected, actual)
+	expectedIssues := map[string][]internal.SchemaIssue{}
+	assert.Equal(t, expectedIssues, conv.Issues[name])
+	// 1 FK issue, 2 index col not found
+	assert.Equal(t, int64(3), conv.Unexpecteds())
+}
+
+func dropComments(t *ddl.CreateTable) {
+	t.Comment = ""
+	for _, c := range t.ColNames {
+		cd := t.ColDefs[c]
+		cd.Comment = ""
+		t.ColDefs[c] = cd
+	}
+}
diff --git a/sources/spanner/infoschema_test.go b/sources/spanner/infoschema_test.go
new file mode 100644
index 0000000..682e2c7
--- /dev/null
+++ b/sources/spanner/infoschema_test.go
@@ -0,0 +1,39 @@
+package spanner
+
+import (
+	"testing"
+
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestToType(t *testing.T) {
+	testCases := []struct {
+		name          string
+		dataType      string
+		expColumnType schema.Type
+	}{
+		// Scalar inputs.
+		{"bool", "BOOL", schema.Type{Name: "BOOL"}},
+		{"int", "INT64", schema.Type{Name: "INT64"}},
+		{"float", "FLOAT64", schema.Type{Name: "FLOAT64"}},
+		{"date", "DATE", schema.Type{Name: "DATE"}},
+		{"numeric", "NUMERIC", schema.Type{Name: "NUMERIC"}},
+		{"json", "JSON", schema.Type{Name: "JSON"}},
+		{"timestamp", "TIMESTAMP", schema.Type{Name: "TIMESTAMP"}},
+		{"bytes", "BYTES(100)", schema.Type{Name: "BYTES", Mods: []int64{100}}},
+		{"bytes", "BYTES(MAX)", schema.Type{Name: "BYTES", Mods: []int64{ddl.MaxLength}}},
+		{"string", "STRING(100)", schema.Type{Name: "STRING", Mods: []int64{100}}},
+		{"string", "STRING(MAX)", schema.Type{Name: "STRING", Mods: []int64{ddl.MaxLength}}},
+		// Array types.
+		{"string_max_arr", "ARRAY<STRING(MAX)>", schema.Type{Name: "STRING", Mods: []int64{ddl.MaxLength}, ArrayBounds: []int64{-1}}},
+		{"string_arr", "ARRAY<STRING(100)>", schema.Type{Name: "STRING", Mods: []int64{100}, ArrayBounds: []int64{-1}}},
+		{"float_arr", "ARRAY<FLOAT64>", schema.Type{Name: "FLOAT64", ArrayBounds: []int64{-1}}},
+		{"numeric_arr", "ARRAY<NUMERIC>", schema.Type{Name: "NUMERIC", ArrayBounds: []int64{-1}}},
+	}
+	for _, tc := range testCases {
+		ty := toType(tc.dataType)
+		assert.Equal(t, tc.expColumnType, ty, tc.name)
+	}
+}
diff --git a/sources/spanner/toddl_test.go b/sources/spanner/toddl_test.go
new file mode 100644
index 0000000..6c0319b
--- /dev/null
+++ b/sources/spanner/toddl_test.go
@@ -0,0 +1,45 @@
+package spanner
+
+import (
+	"testing"
+
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestToSpannerType(t *testing.T) {
+	conv := internal.MakeConv()
+	toDDLImpl := ToDdlImpl{}
+	toDDLTests := []struct {
+		name       string
+		pgTarget   bool
+		columnType schema.Type
+		expDDLType ddl.Type
+	}{
+		// Exact inputs.
+		{"bool", false, schema.Type{Name: "BOOL"}, ddl.Type{Name: ddl.Bool}},
+		{"bytes", false, schema.Type{Name: "BYTES", Mods: []int64{100}}, ddl.Type{Name: ddl.Bytes, Len: 100}},
+		{"date", false, schema.Type{Name: "DATE"}, ddl.Type{Name: ddl.Date}},
+		{"float", false, schema.Type{Name: "FLOAT64"}, ddl.Type{Name: ddl.Float64}},
+		{"int", false, schema.Type{Name: "INT64"}, ddl.Type{Name: ddl.Int64}},
+		{"json", false, schema.Type{Name: "JSON"}, ddl.Type{Name: ddl.JSON}},
+		{"numeric", false, schema.Type{Name: "NUMERIC"}, ddl.Type{Name: ddl.Numeric}},
+		{"string", false, schema.Type{Name: "STRING", Mods: []int64{100}}, ddl.Type{Name: ddl.String, Len: 100}},
+		{"timestamp", false, schema.Type{Name: "TIMESTAMP"}, ddl.Type{Name: ddl.Timestamp}},
+		// PG target.
+		{"pg_numeric", true, schema.Type{Name: "PG.NUMERIC"}, ddl.Type{Name: ddl.Numeric}},
+		{"pg_json", true, schema.Type{Name: "PG.JSONB"}, ddl.Type{Name: ddl.JSON}},
+	}
+	for _, tc := range toDDLTests {
+		conv.TargetDb = constants.TargetSpanner
+		if tc.pgTarget {
+			conv.TargetDb = constants.TargetExperimentalPostgres
+		}
+		ty, err := toDDLImpl.ToSpannerType(conv, tc.columnType)
+		assert.Nil(t, err, tc.name)
+		assert.Equal(t, tc.expDDLType, ty, tc.name)
+	}
+}
diff --git a/spanner/batchwriter_test.go b/spanner/writer/batchwriter_test.go
similarity index 99%
rename from spanner/batchwriter_test.go
rename to spanner/writer/batchwriter_test.go
index 282de76..4f8a503 100644
--- a/spanner/batchwriter_test.go
+++ b/spanner/writer/batchwriter_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package spanner
+package writer
 
 import (
 	"errors"
diff --git a/testing/csv/integration_test.go b/testing/csv/integration_test.go
index a581133..7f43667 100644
--- a/testing/csv/integration_test.go
+++ b/testing/csv/integration_test.go
@@ -48,7 +48,7 @@ var (
 )
 
 const (
-	ALL_TYPES_CSV string = "../../test_data/all_data_types.csv"
+	ALL_TYPES_CSV string = "all_data_types.csv"
 )
 
 type SpannerRecord struct {
@@ -61,6 +61,8 @@ type SpannerRecord struct {
 	AttrString    string
 	AttrTimestamp time.Time
 	AttrJson      spanner.NullJSON
+	AttrStringArr []spanner.NullString
+	AttrInt64Arr  []spanner.NullInt64
 }
 
 func TestMain(m *testing.M) {
@@ -137,6 +139,8 @@ func createSpannerSchema(t *testing.T, project, instance, dbName string) {
 		"g STRING(50)," +
 		"h TIMESTAMP," +
 		"i JSON," +
+		"j ARRAY<STRING(100)>," +
+		"k ARRAY<INT64>," +
 		") PRIMARY KEY(e)",
 	}
 	op, err := databaseAdmin.CreateDatabase(ctx, req)
@@ -157,12 +161,11 @@ func TestIntegration_CSV_Command(t *testing.T) {
 
 	dbName := "csv-test"
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	manifest := "../../test_data/csv_manifest.json"
 
 	writeCSVs(t)
 	defer cleanupCSVs()
 	createSpannerSchema(t, projectID, instanceID, dbName)
-	args := fmt.Sprintf("data -source=csv -source-profile='manifest=%s' -target-profile='instance=%s,dbname=%s'", manifest, instanceID, dbName)
+	args := fmt.Sprintf("data -source=csv -target-profile='instance=%s,dbname=%s'", instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -181,8 +184,8 @@ func writeCSVs(t *testing.T) {
 		{
 			ALL_TYPES_CSV,
 			[]string{
-				"a,b,c,d,e,f,g,h,i\n",
-				"true,test,2019-10-29,15.13,100,39.94,Helloworld,2019-10-29 05:30:00,\"{\"\"key1\"\": \"\"value1\"\", \"\"key2\"\": \"\"value2\"\"}\"",
+				"a,b,c,d,e,f,g,h,i,j,k\n",
+				"true,test,2019-10-29,15.13,100,39.94,Helloworld,2019-10-29 05:30:00,\"{\"\"key1\"\": \"\"value1\"\", \"\"key2\"\": \"\"value2\"\"}\",\"{ab,cd}\",\"[1,2]\"",
 			},
 		},
 	}
@@ -225,11 +228,13 @@ func checkRow(ctx context.Context, t *testing.T, client *spanner.Client) {
 		AttrString:    "Helloworld",
 		AttrTimestamp: getTime(t, "2019-10-29T05:30:00Z"),
 		AttrJson:      spanner.NullJSON{Valid: true},
+		AttrStringArr: []spanner.NullString{{StringVal: "ab", Valid: true}, {StringVal: "cd", Valid: true}},
+		AttrInt64Arr:  []spanner.NullInt64{{Int64: int64(1), Valid: true}, {Int64: int64(2), Valid: true}},
 	}
 	json.Unmarshal([]byte("{\"key1\": \"value1\", \"key2\": \"value2\"}"), &wantRecord.AttrJson.Value)
 
 	gotRecord := SpannerRecord{}
-	stmt := spanner.Statement{SQL: `SELECT a, b, c, d, e, f, g, h, i FROM all_data_types`}
+	stmt := spanner.Statement{SQL: `SELECT a, b, c, d, e, f, g, h, i, j, k FROM all_data_types`}
 	iter := client.Single().Query(ctx, stmt)
 	defer iter.Stop()
 	for {
@@ -245,7 +250,7 @@ func checkRow(ctx context.Context, t *testing.T, client *spanner.Client) {
 		// We don't create big.Rat fields in the SpannerRecord structs
 		// because cmp.Equal cannot compare big.Rat fields automatically.
 		var AttrNumeric big.Rat
-		if err := row.Columns(&gotRecord.AttrBool, &gotRecord.AttrBytes, &gotRecord.AttrDate, &gotRecord.AttrFloat, &gotRecord.AttrInt, &AttrNumeric, &gotRecord.AttrString, &gotRecord.AttrTimestamp, &gotRecord.AttrJson); err != nil {
+		if err := row.Columns(&gotRecord.AttrBool, &gotRecord.AttrBytes, &gotRecord.AttrDate, &gotRecord.AttrFloat, &gotRecord.AttrInt, &AttrNumeric, &gotRecord.AttrString, &gotRecord.AttrTimestamp, &gotRecord.AttrJson, &gotRecord.AttrStringArr, &gotRecord.AttrInt64Arr); err != nil {
 			log.Println("Error reading into variables: ", err)
 			t.Fatal(err)
 			break