Repository: cloudspannerecosystem/harbourbridge (go)

Issue:
Standardizing DynamoDb implementation to use common structure
## Expected Behavior
The DynamoDb implementation should use the same interfaces as the common db code in sources/common/ as outlined in Issue #178 and PR #181 

## Actual Behavior
Current implementation is completely independent of the structure used in Mysql and Postgres.
Making it follow the same code flow will ensure a more standard code base and provide a template for further non-relational DB integrations.

Pull request:
Remove sql dependency from common
Fixes #189 
These changes aim to remove the sql dependency from common package to make is usable for dynamodb (or any other non sql database) implementation. It includes the following changes:
1. Merging GetColumns and ProcessColumns together.
Mysql and Postgres get the columns names first and then fetch their definitions. However, for dynamo, sample data is scanned and column names and definitions are derived from that together. Merging these 2 methods would make the common implementation useful for dynamo as well.
2. Remove the sql data rows specific logic from common and refactor ProcessData to make it useful for Dynamodb implementation.
3. Move sql.db inside InfoSchema structs.
Mysql and Postgres use sql.db client while dynamo uses dydb client library. After the above 2 changes, sql.db usage in common ceases to exist so there is no need to keep it in common just to pass it down as method arguments to mysql/postgres infoschema implementations.


- [x] Tests pass
- [x] Appropriate changes to README are included in PR

Patch:
diff --git a/conversion/conversion.go b/conversion/conversion.go
index d20292e..dc7e12c 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -202,7 +202,7 @@ func schemaFromSQL(driver, sqlConnectionStr, targetDb string) (*internal.Conv, e
 	conv := internal.MakeConv()
 	conv.TargetDb = targetDb
 	conv.SrcDbName = getDbNameFromSQLConnectionStr(driver, sqlConnectionStr)
-	err = ProcessInfoSchema(driver, conv, sourceDB)
+	err = ProcessSchema(driver, conv, sourceDB)
 	if err != nil {
 		return nil, err
 	}
@@ -249,7 +249,7 @@ func dataFromSQL(driver, sqlConnectionStr string, config spanner.BatchWriterConf
 		func(table string, cols []string, vals []interface{}) {
 			writer.AddRow(table, cols, vals)
 		})
-	err = ProcessSQLData(driver, conv, sourceDB)
+	err = ProcessData(driver, conv, sourceDB)
 	if err != nil {
 		return nil, err
 	}
@@ -1158,13 +1158,13 @@ func ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 	}
 }
 
-// ProcessInfoSchema invokes process infoschema function from a sql package based on driver selected.
-func ProcessInfoSchema(driver string, conv *internal.Conv, db *sql.DB) error {
+// ProcessSchema invokes processSchema function from a sql package based on driver selected.
+func ProcessSchema(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
 	case constants.MYSQL:
-		return common.ProcessInfoSchema(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
+		return common.ProcessSchema(conv, mysql.InfoSchemaImpl{DbName: conv.SrcDbName, Db: db})
 	case constants.POSTGRES:
-		return common.ProcessInfoSchema(conv, db, postgres.InfoSchemaImpl{})
+		return common.ProcessSchema(conv, postgres.InfoSchemaImpl{Db: db})
 	default:
 		return fmt.Errorf("schema conversion for driver %s not supported", driver)
 	}
@@ -1174,23 +1174,23 @@ func ProcessInfoSchema(driver string, conv *internal.Conv, db *sql.DB) error {
 func SetRowStats(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
 	case constants.MYSQL:
-		common.SetRowStats(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
+		common.SetRowStats(conv, mysql.InfoSchemaImpl{DbName: conv.SrcDbName, Db: db})
 	case constants.POSTGRES:
-		common.SetRowStats(conv, db, postgres.InfoSchemaImpl{})
+		common.SetRowStats(conv, postgres.InfoSchemaImpl{Db: db})
 	default:
 		return fmt.Errorf("could not set rows stats for '%s' driver", driver)
 	}
 	return nil
 }
 
-// ProcessSQLData invokes ProcessSQLData function from a sql package based on driver selected.
-func ProcessSQLData(driver string, conv *internal.Conv, db *sql.DB) error {
+// ProcessData invokes ProcessData function from a sql package based on driver selected.
+func ProcessData(driver string, conv *internal.Conv, db *sql.DB) error {
 	switch driver {
 	//TODO - move this logic into a factory within the sources dir
 	case constants.MYSQL:
-		common.ProcessSQLData(conv, db, mysql.InfoSchemaImpl{conv.SrcDbName})
+		common.ProcessData(conv, mysql.InfoSchemaImpl{DbName: conv.SrcDbName, Db: db})
 	case constants.POSTGRES:
-		common.ProcessSQLData(conv, db, postgres.InfoSchemaImpl{})
+		common.ProcessData(conv, postgres.InfoSchemaImpl{Db: db})
 	default:
 		return fmt.Errorf("data conversion for driver %s is not supported", driver)
 	}
diff --git a/schema/schema.go b/schema/schema.go
index 41a22b0..49f10b2 100644
--- a/schema/schema.go
+++ b/schema/schema.go
@@ -34,6 +34,7 @@ import (
 // Table represents a database table.
 type Table struct {
 	Name        string
+	Schema      string
 	ColNames    []string          // List of column names (for predictable iteration order e.g. printing).
 	ColDefs     map[string]Column // Details of columns.
 	PrimaryKeys []Key
diff --git a/sources/common/infoschema.go b/sources/common/infoschema.go
index 9b5ac42..320dcca 100644
--- a/sources/common/infoschema.go
+++ b/sources/common/infoschema.go
@@ -15,7 +15,6 @@
 package common
 
 import (
-	"database/sql"
 	"fmt"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -27,15 +26,14 @@ import (
 type InfoSchema interface {
 	GetToDdl() ToDdl
 	GetTableName(schema string, tableName string) string
-	GetTables(db *sql.DB) ([]SchemaAndName, error)
-	GetColumns(table SchemaAndName, db *sql.DB) (*sql.Rows, error) //TODO - merge this method and ProcessColumns for cleaner interface
-	ProcessColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string)
-	GetRowsFromTable(conv *internal.Conv, db *sql.DB, table SchemaAndName) (*sql.Rows, error)
-	GetRowCount(db *sql.DB, table SchemaAndName) (int64, error)
-	GetConstraints(conv *internal.Conv, db *sql.DB, table SchemaAndName) ([]string, map[string][]string, error)
-	GetForeignKeys(conv *internal.Conv, db *sql.DB, table SchemaAndName) (foreignKeys []schema.ForeignKey, err error)
-	GetIndexes(conv *internal.Conv, db *sql.DB, table SchemaAndName) ([]schema.Index, error)
-	ProcessDataRows(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, rows *sql.Rows)
+	GetTables() ([]SchemaAndName, error)
+	GetColumns(conv *internal.Conv, table SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error)
+	GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error)
+	GetRowCount(table SchemaAndName) (int64, error)
+	GetConstraints(conv *internal.Conv, table SchemaAndName) ([]string, map[string][]string, error)
+	GetForeignKeys(conv *internal.Conv, table SchemaAndName) (foreignKeys []schema.ForeignKey, err error)
+	GetIndexes(conv *internal.Conv, table SchemaAndName) ([]schema.Index, error)
+	ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error
 }
 
 // SchemaAndName contains the schema and name for a table
@@ -52,16 +50,16 @@ type FkConstraint struct {
 	Cols    []string
 }
 
-// ProcessInfoSchema performs schema conversion for source database
+// ProcessSchema performs schema conversion for source database
 // 'db'. Information schema tables are a broadly supported ANSI standard,
 // and we use them to obtain source database's schema information.
-func ProcessInfoSchema(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) error {
-	tables, err := infoSchema.GetTables(db)
+func ProcessSchema(conv *internal.Conv, infoSchema InfoSchema) error {
+	tables, err := infoSchema.GetTables()
 	if err != nil {
 		return err
 	}
 	for _, t := range tables {
-		if err := processTable(conv, db, t, infoSchema); err != nil {
+		if err := processTable(conv, t, infoSchema); err != nil {
 			return err
 		}
 	}
@@ -70,68 +68,39 @@ func ProcessInfoSchema(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) e
 	return nil
 }
 
-// ProcessSQLData performs data conversion for source database
+// ProcessData performs data conversion for source database
 // 'db'. For each table, we extract and convert the data to Spanner data
 // (based on the source and Spanner schemas), and write it to Spanner.
 // If we can't get/process data for a table, we skip that table and process
 // the remaining tables.
-//
-// Using database/sql library we pass *sql.RawBytes to rows.scan.
-// RawBytes is a byte slice and values can be easily converted to string.
-func ProcessSQLData(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) {
-	// TODO: refactor to use the set of tables computed by
-	// ProcessInfoSchema instead of computing them again.
-	tables, err := infoSchema.GetTables(db)
-	if err != nil {
-		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
-		return
-	}
-	for _, t := range tables {
-		srcTable := infoSchema.GetTableName(t.Schema, t.Name)
-		srcSchema, ok := conv.SrcSchema[srcTable]
-		if !ok {
+func ProcessData(conv *internal.Conv, infoSchema InfoSchema) {
+	for srcTable, srcSchema := range conv.SrcSchema {
+		spTable, err1 := internal.GetSpannerTable(conv, srcTable)
+		spCols, err2 := internal.GetSpannerCols(conv, srcTable, srcSchema.ColNames)
+		spSchema, ok := conv.SpSchema[spTable]
+		if err1 != nil || err2 != nil || !ok {
 			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't get schemas for table %s", srcTable))
-			continue
-		}
-		rows, err := infoSchema.GetRowsFromTable(conv, db, t)
-		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", t.Name, err))
-			continue
-		}
-		defer rows.Close()
-		srcCols, _ := rows.Columns()
-		spTable, err := internal.GetSpannerTable(conv, srcTable)
-		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Couldn't get spanner table : %s", err))
+			conv.Unexpected(fmt.Sprintf("Can't get cols and schemas for table %s: err1=%s, err2=%s, ok=%t",
+				srcTable, err1, err2, ok))
 			continue
 		}
-		spCols, err := internal.GetSpannerCols(conv, srcTable, srcCols)
+		err := infoSchema.ProcessData(conv, srcTable, srcSchema, spTable, spCols, spSchema)
 		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Couldn't get spanner columns for table %s : err = %s", t.Name, err))
-			continue
+			return
 		}
-		spSchema, ok := conv.SpSchema[spTable]
-		if !ok {
-			//TODO - check why Bad rows are not being added in above conditions
-			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
-			conv.Unexpected(fmt.Sprintf("Can't get schemas for table %s", srcTable))
-			continue
-		}
-		infoSchema.ProcessDataRows(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, rows)
 	}
 }
 
 // SetRowStats populates conv with the number of rows in each table.
-func SetRowStats(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) {
-	tables, err := infoSchema.GetTables(db)
+func SetRowStats(conv *internal.Conv, infoSchema InfoSchema) {
+	tables, err := infoSchema.GetTables()
 	if err != nil {
 		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
 		return
 	}
 	for _, t := range tables {
 		tableName := infoSchema.GetTableName(t.Schema, t.Name)
-		count, err := infoSchema.GetRowCount(db, t)
+		count, err := infoSchema.GetRowCount(t)
 		if err != nil {
 			conv.Unexpected(fmt.Sprintf("Couldn't get number of rows for table %s", tableName))
 			continue
@@ -140,26 +109,20 @@ func SetRowStats(conv *internal.Conv, db *sql.DB, infoSchema InfoSchema) {
 	}
 }
 
-func processTable(conv *internal.Conv, db *sql.DB, table SchemaAndName, infoSchema InfoSchema) error {
-	cols, err := infoSchema.GetColumns(table, db)
-	if err != nil {
-		return fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
-	}
-	defer cols.Close()
-
-	primaryKeys, constraints, err := infoSchema.GetConstraints(conv, db, table)
+func processTable(conv *internal.Conv, table SchemaAndName, infoSchema InfoSchema) error {
+	primaryKeys, constraints, err := infoSchema.GetConstraints(conv, table)
 	if err != nil {
 		return fmt.Errorf("couldn't get constraints for table %s.%s: %s", table.Schema, table.Name, err)
 	}
-	foreignKeys, err := infoSchema.GetForeignKeys(conv, db, table)
+	foreignKeys, err := infoSchema.GetForeignKeys(conv, table)
 	if err != nil {
 		return fmt.Errorf("couldn't get foreign key constraints for table %s.%s: %s", table.Schema, table.Name, err)
 	}
-	indexes, err := infoSchema.GetIndexes(conv, db, table)
+	indexes, err := infoSchema.GetIndexes(conv, table)
 	if err != nil {
 		return fmt.Errorf("couldn't get indexes for table %s.%s: %s", table.Schema, table.Name, err)
 	}
-	colDefs, colNames := infoSchema.ProcessColumns(conv, cols, constraints)
+	colDefs, colNames, err := infoSchema.GetColumns(conv, table, constraints, primaryKeys)
 	if err != nil {
 		return fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
 	}
@@ -170,6 +133,7 @@ func processTable(conv *internal.Conv, db *sql.DB, table SchemaAndName, infoSche
 	}
 	conv.SrcSchema[name] = schema.Table{
 		Name:        name,
+		Schema:      table.Schema,
 		ColNames:    colNames,
 		ColDefs:     colDefs,
 		PrimaryKeys: schemaPKeys,
diff --git a/sources/mysql/infoschema.go b/sources/mysql/infoschema.go
index 47a016a..95bf33a 100644
--- a/sources/mysql/infoschema.go
+++ b/sources/mysql/infoschema.go
@@ -31,6 +31,7 @@ import (
 // InfoSchemaImpl is MySQL specific implementation for InfoSchema.
 type InfoSchemaImpl struct {
 	DbName string
+	Db     *sql.DB
 }
 
 // GetToDdl implement the common.InfoSchema interface.
@@ -44,19 +45,19 @@ func (isi InfoSchemaImpl) GetTableName(dbName string, tableName string) string {
 }
 
 // GetRowsFromTable returns a sql Rows object for a table.
-func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (*sql.Rows, error) {
-	srcSchema := conv.SrcSchema[table.Name]
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
+	srcSchema := conv.SrcSchema[srcTable]
 	srcCols := srcSchema.ColNames
 	if len(srcCols) == 0 {
-		conv.Unexpected(fmt.Sprintf("Couldn't get source columns for table %s ", table.Name))
+		conv.Unexpected(fmt.Sprintf("Couldn't get source columns for table %s ", srcTable))
 		return nil, nil
 	}
 	// MySQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but MySQL doesn't support this. So we quote it instead.
 	colNameList := buildColNameList(srcSchema, srcCols)
-	q := fmt.Sprintf("SELECT %s FROM `%s`.`%s`;", colNameList, table.Schema, table.Name)
-	rows, err := db.Query(q)
+	q := fmt.Sprintf("SELECT %s FROM `%s`.`%s`;", colNameList, conv.SrcSchema[srcTable].Schema, srcTable)
+	rows, err := isi.Db.Query(q)
 	return rows, err
 }
 
@@ -80,8 +81,16 @@ func buildColNameList(srcSchema schema.Table, srcColName []string) string {
 	return colList[:len(colList)-1]
 }
 
-// ProcessDataRows performs data conversion for source database.
-func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, rows *sql.Rows) {
+// ProcessData performs data conversion for source database.
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	rowsInterface, err := isi.GetRowsFromTable(conv, srcTable)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", srcTable, err))
+		return err
+	}
+	rows := rowsInterface.(*sql.Rows)
+	defer rows.Close()
+	srcCols, _ := rows.Columns()
 	v, scanArgs := buildVals(len(srcCols))
 	for rows.Next() {
 		// get RawBytes from data.
@@ -95,15 +104,16 @@ func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string,
 		values := valsToStrings(v)
 		ProcessDataRow(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, values)
 	}
+	return nil
 }
 
 // GetRowCount with number of rows in each table.
-func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (int64, error) {
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
 	// MySQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but MySQL doesn't support this. So we quote it instead.
 	q := fmt.Sprintf("SELECT COUNT(*) FROM `%s`.`%s`;", table.Schema, table.Name)
-	rows, err := db.Query(q)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return 0, err
 	}
@@ -120,10 +130,10 @@ func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (i
 // Note that sql.DB already effectively has the dbName
 // embedded within it (dbName is part of the DSN passed to sql.Open),
 // but unfortunately there is no way to extract it from sql.DB.
-func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error) {
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
 	// In MySQL, schema is the same as database name.
 	q := "SELECT table_name FROM information_schema.tables where table_type = 'BASE TABLE' and table_schema=?"
-	rows, err := db.Query(q, isi.DbName)
+	rows, err := isi.Db.Query(q, isi.DbName)
 	if err != nil {
 		return nil, fmt.Errorf("couldn't get tables: %w", err)
 	}
@@ -137,16 +147,15 @@ func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error)
 	return tables, nil
 }
 
-// GetColumns returns a list of columns with their data type
-func (isi InfoSchemaImpl) GetColumns(table common.SchemaAndName, db *sql.DB) (*sql.Rows, error) {
+// GetColumns returns a list of Column objects and names// ProcessColumns
+func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error) {
 	q := `SELECT c.column_name, c.data_type, c.column_type, c.is_nullable, c.column_default, c.character_maximum_length, c.numeric_precision, c.numeric_scale, c.extra
               FROM information_schema.COLUMNS c
               where table_schema = ? and table_name = ? ORDER BY c.ordinal_position;`
-	return db.Query(q, table.Schema, table.Name)
-}
-
-// ProcessColumns returns a list of Column objects and names// ProcessColumns
-func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string) {
+	cols, err := isi.Db.Query(q, table.Schema, table.Name)
+	if err != nil {
+		return nil, nil, fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
+	}
 	colDefs := make(map[string]schema.Column)
 	var colNames []string
 	var colName, dataType, isNullable, columnType string
@@ -182,20 +191,20 @@ func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, co
 		colDefs[colName] = c
 		colNames = append(colNames, colName)
 	}
-	return colDefs, colNames
+	return colDefs, colNames, nil
 }
 
 // GetConstraints returns a list of primary keys and by-column map of
 // other constraints.  Note: we need to preserve ordinal order of
 // columns in primary key constraints.
 // Note that foreign key constraints are handled in getForeignKeys.
-func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]string, map[string][]string, error) {
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) ([]string, map[string][]string, error) {
 	q := `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
               FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
                 INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
                   ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA AND t.TABLE_NAME=k.TABLE_NAME
               WHERE k.TABLE_SCHEMA = ? AND k.TABLE_NAME = ? ORDER BY k.ordinal_position;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -228,7 +237,7 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table
 // them because HarbourBridge works database at a time (a specific run
 // of HarbourBridge focuses on a specific database) and so we can't handle
 // them effectively.
-func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
 	q := `SELECT k.REFERENCED_TABLE_NAME,k.COLUMN_NAME,k.REFERENCED_COLUMN_NAME,k.CONSTRAINT_NAME
 		FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t 
 		INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k 
@@ -243,7 +252,7 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
 			k.REFERENCED_TABLE_NAME,
 			k.COLUMN_NAME,
 			k.ORDINAL_POSITION;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
@@ -281,14 +290,14 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
 }
 
 // GetIndexes return a list of all indexes for the specified table.
-func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]schema.Index, error) {
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
 	q := `SELECT DISTINCT INDEX_NAME,COLUMN_NAME,SEQ_IN_INDEX,COLLATION,NON_UNIQUE
 		FROM INFORMATION_SCHEMA.STATISTICS 
 		WHERE TABLE_SCHEMA = ?
 			AND TABLE_NAME = ?
 			AND INDEX_NAME != 'PRIMARY' 
 		ORDER BY INDEX_NAME, SEQ_IN_INDEX;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
diff --git a/sources/postgres/infoschema.go b/sources/postgres/infoschema.go
index 7ea1f19..cfea57a 100644
--- a/sources/postgres/infoschema.go
+++ b/sources/postgres/infoschema.go
@@ -34,6 +34,7 @@ import (
 
 // InfoSchemaImpl postgres specific implementation for InfoSchema.
 type InfoSchemaImpl struct {
+	Db *sql.DB
 }
 
 // GetToDdl function below implement the common.InfoSchema interface.
@@ -50,12 +51,12 @@ func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
 }
 
 // GetRowsFromTable returns a sql Rows object for a table.
-func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (*sql.Rows, error) {
+func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
 	// PostgreSQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but PostgreSQL doesn't support this. So we quote it instead.
-	q := fmt.Sprintf(`SELECT * FROM "%s"."%s";`, table.Schema, table.Name)
-	rows, err := db.Query(q)
+	q := fmt.Sprintf(`SELECT * FROM "%s"."%s";`, conv.SrcSchema[srcTable].Schema, srcTable)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return nil, err
 	}
@@ -81,7 +82,15 @@ func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, db *sql.DB, tabl
 // We choose to do all type conversions explicitly ourselves so that
 // we can generate more targeted error messages: hence we pass
 // *interface{} parameters to row.Scan.
-func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, rows *sql.Rows) {
+func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable) error {
+	rowsInterface, err := isi.GetRowsFromTable(conv, srcTable)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get data for table %s : err = %s", srcTable, err))
+		return err
+	}
+	rows := rowsInterface.(*sql.Rows)
+	defer rows.Close()
+	srcCols, _ := rows.Columns()
 	v, iv := buildVals(len(srcCols))
 	for rows.Next() {
 		err := rows.Scan(iv...)
@@ -100,6 +109,7 @@ func (isi InfoSchemaImpl) ProcessDataRows(conv *internal.Conv, srcTable string,
 		}
 		conv.WriteRow(srcTable, spTable, cvtCols, cvtVals)
 	}
+	return nil
 }
 
 // ConvertSQLRow performs data conversion for a single row of data
@@ -143,12 +153,12 @@ func convertSQLRow(conv *internal.Conv, srcTable string, srcCols []string, srcSc
 }
 
 // GetRowCount with number of rows in each table.
-func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (int64, error) {
+func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
 	// PostgreSQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but PostgreSQL doesn't support this. So we quote it instead.
 	q := fmt.Sprintf(`SELECT COUNT(*) FROM "%s"."%s";`, table.Schema, table.Name)
-	rows, err := db.Query(q)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return 0, err
 	}
@@ -166,14 +176,14 @@ func (isi InfoSchemaImpl) GetRowCount(db *sql.DB, table common.SchemaAndName) (i
 // a single transaction to ensure we obtain a consistent snapshot of
 // schema information and table data (pg_dump does something
 // similar).
-func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error) {
+func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
 	ignored := make(map[string]bool)
 	// Ignore all system tables: we just want to convert user tables.
 	for _, s := range []string{"information_schema", "postgres", "pg_catalog", "pg_temp_1", "pg_toast", "pg_toast_temp_1"} {
 		ignored[s] = true
 	}
 	q := "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'"
-	rows, err := db.Query(q)
+	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return nil, fmt.Errorf("couldn't get tables: %w", err)
 	}
@@ -189,18 +199,17 @@ func (isi InfoSchemaImpl) GetTables(db *sql.DB) ([]common.SchemaAndName, error)
 	return tables, nil
 }
 
-// GetColumns returns a list of columns with their data type
-func (isi InfoSchemaImpl) GetColumns(table common.SchemaAndName, db *sql.DB) (*sql.Rows, error) {
+// GetColumns returns a list of Column objects and names
+func (isi InfoSchemaImpl) GetColumns(conv *internal.Conv, table common.SchemaAndName, constraints map[string][]string, primaryKeys []string) (map[string]schema.Column, []string, error) {
 	q := `SELECT c.column_name, c.data_type, e.data_type, c.is_nullable, c.column_default, c.character_maximum_length, c.numeric_precision, c.numeric_scale
               FROM information_schema.COLUMNS c LEFT JOIN information_schema.element_types e
                  ON ((c.table_catalog, c.table_schema, c.table_name, 'TABLE', c.dtd_identifier)
                      = (e.object_catalog, e.object_schema, e.object_name, e.object_type, e.collection_type_identifier))
               where table_schema = $1 and table_name = $2 ORDER BY c.ordinal_position;`
-	return db.Query(q, table.Schema, table.Name)
-}
-
-// ProcessColumns returns a list of Column objects and names
-func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string) {
+	cols, err := isi.Db.Query(q, table.Schema, table.Name)
+	if err != nil {
+		return nil, nil, fmt.Errorf("couldn't get schema for table %s.%s: %s", table.Schema, table.Name, err)
+	}
 	colDefs := make(map[string]schema.Column)
 	var colNames []string
 	var colName, dataType, isNullable string
@@ -234,20 +243,20 @@ func (isi InfoSchemaImpl) ProcessColumns(conv *internal.Conv, cols *sql.Rows, co
 		colDefs[colName] = c
 		colNames = append(colNames, colName)
 	}
-	return colDefs, colNames
+	return colDefs, colNames, nil
 }
 
 // GetConstraints returns a list of primary keys and by-column map of
 // other constraints.  Note: we need to preserve ordinal order of
 // columns in primary key constraints.
 // Note that foreign key constraints are handled in getForeignKeys.
-func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]string, map[string][]string, error) {
+func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.SchemaAndName) ([]string, map[string][]string, error) {
 	q := `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
               FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
                 INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
                   ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
               WHERE k.TABLE_SCHEMA = $1 AND k.TABLE_NAME = $2 ORDER BY k.ordinal_position;`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -276,7 +285,7 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, db *sql.DB, table
 }
 
 // GetForeignKeys returns a list of all the foreign key constraints.
-func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
+func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
 	q := `SELECT 
 		schema_name AS "TABLE_SCHEMA", 
 		cl.relname AS "TABLE_NAME", 
@@ -301,7 +310,7 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
    		JOIN PG_ATTRIBUTE att2 ON
        		att2.attrelid = con.conrelid AND att2.attnum = con.parent;`
 
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
@@ -344,7 +353,7 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, db *sql.DB, table
 // Note: Extracting index definitions from PostgreSQL information schema tables is complex.
 // See https://stackoverflow.com/questions/6777456/list-all-index-names-column-names-and-its-table-name-of-a-postgresql-database/44460269#44460269
 // for background.
-func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, db *sql.DB, table common.SchemaAndName) ([]schema.Index, error) {
+func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
 	q := `SELECT
 			irel.relname AS index_name,
 			a.attname AS column_name,
@@ -374,7 +383,7 @@ func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, db *sql.DB, table comm
            		array_position(i.indkey, a.attnum),
            		o.OPTION,i.indisunique
 		ORDER BY irel.relname, array_position(i.indkey, a.attnum);`
-	rows, err := db.Query(q, table.Schema, table.Name)
+	rows, err := isi.Db.Query(q, table.Schema, table.Name)
 	if err != nil {
 		return nil, err
 	}
diff --git a/web/web.go b/web/web.go
index decded0..8a2a882 100644
--- a/web/web.go
+++ b/web/web.go
@@ -125,9 +125,9 @@ func convertSchemaSQL(w http.ResponseWriter, r *http.Request) {
 	var err error
 	switch sessionState.driver {
 	case constants.MYSQL:
-		err = common.ProcessInfoSchema(conv, sessionState.sourceDB, mysql.InfoSchemaImpl{sessionState.dbName})
+		err = common.ProcessSchema(conv, mysql.InfoSchemaImpl{DbName: sessionState.dbName, Db: sessionState.sourceDB})
 	case constants.POSTGRES:
-		err = common.ProcessInfoSchema(conv, sessionState.sourceDB, postgres.InfoSchemaImpl{})
+		err = common.ProcessSchema(conv, postgres.InfoSchemaImpl{Db: sessionState.sourceDB})
 	default:
 		http.Error(w, fmt.Sprintf("Driver : '%s' is not supported", sessionState.driver), http.StatusBadRequest)
 		return


Test patch:
diff --git a/sources/mysql/data_test.go b/sources/mysql/data_test.go
index 55223e2..ebe3937 100644
--- a/sources/mysql/data_test.go
+++ b/sources/mysql/data_test.go
@@ -320,12 +320,6 @@ func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *internal.Conv {
 	conv := internal.MakeConv()
 	conv.SpSchema[spTable.Name] = spTable
 	conv.SrcSchema[srcTable.Name] = srcTable
-	conv.ToSource[spTable.Name] = internal.NameAndCols{Name: srcTable.Name, Cols: make(map[string]string)}
-	conv.ToSpanner[srcTable.Name] = internal.NameAndCols{Name: spTable.Name, Cols: make(map[string]string)}
-	for i := range spTable.ColNames {
-		conv.ToSource[spTable.Name].Cols[spTable.ColNames[i]] = srcTable.ColNames[i]
-		conv.ToSpanner[srcTable.Name].Cols[srcTable.ColNames[i]] = spTable.ColNames[i]
-	}
 	return conv
 }
 
diff --git a/sources/mysql/infoschema_test.go b/sources/mysql/infoschema_test.go
index 0c56fe2..0f1c841 100644
--- a/sources/mysql/infoschema_test.go
+++ b/sources/mysql/infoschema_test.go
@@ -34,7 +34,7 @@ type mockSpec struct {
 	rows  [][]driver.Value // Set of rows returned.
 }
 
-func TestProcessInfoSchemaMYSQL(t *testing.T) {
+func TestProcessSchemaMYSQL(t *testing.T) {
 	ms := []mockSpec{
 
 		{
@@ -47,14 +47,6 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"product"},
 				{"test"},
 				{"test_ref"}},
-		}, {
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "user"},
-			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
-			rows: [][]driver.Value{
-				{"user_id", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"name", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"ref", "bigint", "bigint", "NO", nil, nil, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "user"},
@@ -73,16 +65,16 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
 			args:  []driver.Value{"test", "user"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
-		},
-		{
+		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "cart"},
+			args:  []driver.Value{"test", "user"},
 			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
 			rows: [][]driver.Value{
-				{"productid", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"userid", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"quantity", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
-		}, {
+				{"user_id", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"name", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"ref", "bigint", "bigint", "NO", nil, nil, nil, nil, nil}},
+		},
+		{
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "cart"},
 			cols:  []string{"column_name", "constraint_type"},
@@ -108,11 +100,12 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"index3", "userid", 2, "D", "0"}},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "product"},
+			args:  []driver.Value{"test", "cart"},
 			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
 			rows: [][]driver.Value{
-				{"product_id", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"product_name", "text", "text", "NO", nil, nil, nil, nil, nil}},
+				{"productid", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"userid", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"quantity", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "product"},
@@ -127,6 +120,28 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
 			args:  []driver.Value{"test", "product"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"test", "product"},
+			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
+			rows: [][]driver.Value{
+				{"product_id", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"product_name", "text", "text", "NO", nil, nil, nil, nil, nil}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}, {"id", "FOREIGN KEY"}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
+			rows: [][]driver.Value{{"test_ref", "id", "ref_id", "fk_test4"},
+				{"test_ref", "txt", "ref_txt", "fk_test4"}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"test", "test"},
@@ -154,18 +169,18 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"vc6", "varchar", "varchar(6)", "YES", nil, 6, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test"},
+			args:  []driver.Value{"test", "test_ref"},
 			cols:  []string{"column_name", "constraint_type"},
-			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}, {"id", "FOREIGN KEY"}},
+			rows: [][]driver.Value{
+				{"ref_id", "PRIMARY KEY"},
+				{"ref_txt", "PRIMARY KEY"}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test"},
+			args:  []driver.Value{"test", "test_ref"},
 			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
-			rows: [][]driver.Value{{"test_ref", "id", "ref_id", "fk_test4"},
-				{"test_ref", "txt", "ref_txt", "fk_test4"}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
-			args:  []driver.Value{"test", "test"},
+			args:  []driver.Value{"test", "test_ref"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
@@ -175,26 +190,11 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 				{"ref_id", "bigint", "bigint", "NO", nil, nil, 64, 0, nil},
 				{"ref_txt", "text", "text", "NO", nil, nil, nil, nil, nil},
 				{"abc", "text", "text", "NO", nil, nil, nil, nil, nil}},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test_ref"},
-			cols:  []string{"column_name", "constraint_type"},
-			rows: [][]driver.Value{
-				{"ref_id", "PRIMARY KEY"},
-				{"ref_txt", "PRIMARY KEY"}},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"test", "test_ref"},
-			cols:  []string{"REFERENCED_TABLE_NAME", "COLUMN_NAME", "REFERENCED_COLUMN_NAME", "CONSTRAINT_NAME"},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
-			args:  []driver.Value{"test", "test_ref"},
-			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
 		},
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{"test"})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"user": ddl.CreateTable{
@@ -280,14 +280,9 @@ func TestProcessInfoSchemaMYSQL(t *testing.T) {
 	assert.Equal(t, int64(0), conv.Unexpecteds())
 }
 
-func TestProcessSQLData(t *testing.T) {
+func TestProcessData(t *testing.T) {
 	ms := []mockSpec{
 		{
-			query: "SELECT table_name FROM information_schema.tables where table_type = 'BASE TABLE' and (.+)",
-			args:  []driver.Value{"test"},
-			cols:  []string{"table_name"},
-			rows:  [][]driver.Value{{"te st"}},
-		}, {
 			query: "SELECT (.+) FROM `test`.`te st`",
 			cols:  []string{"a a", " b", " c "},
 			rows: [][]driver.Value{
@@ -308,6 +303,7 @@ func TestProcessSQLData(t *testing.T) {
 			}},
 		schema.Table{
 			Name:     "te st",
+			Schema:   "test",
 			ColNames: []string{"a_a", "_b", "_c_"},
 			ColDefs: map[string]schema.Column{
 				"a a": schema.Column{Name: "a a", Type: schema.Type{Name: "float"}},
@@ -321,7 +317,7 @@ func TestProcessSQLData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{"test"})
+	common.ProcessData(conv, InfoSchemaImpl{"test", db})
 	assert.Equal(t,
 		[]spannerData{
 			spannerData{table: "te_st", cols: []string{"a_a", "Ab", "Ac_"}, vals: []interface{}{float64(42.3), int64(3), "cat"}},
@@ -333,7 +329,7 @@ func TestProcessSQLData(t *testing.T) {
 	assert.Equal(t, int64(1), conv.Unexpecteds()) // Bad row generates an entry in unexpected.
 }
 
-func TestProcessSQLData_MultiCol(t *testing.T) {
+func TestProcessData_MultiCol(t *testing.T) {
 	// Tests multi-column behavior of ProcessSQLData (including
 	// handling of null columns and synthetic keys). Also tests
 	// the combination of ProcessInfoSchema and ProcessSQLData
@@ -346,15 +342,6 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 			cols:  []string{"table_name"},
 			rows:  [][]driver.Value{{"test"}},
 		}, {
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"test", "test"},
-			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
-			rows: [][]driver.Value{
-				{"a", "text", "text", "NO", nil, nil, nil, nil, nil},
-				{"b", "double", "double", "YES", nil, nil, 53, nil, nil},
-				{"c", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
-		},
-		{
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"test", "test"},
 			cols:  []string{"column_name", "constraint_type"},
@@ -367,17 +354,16 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.STATISTICS (.+)",
 			args:  []driver.Value{"test", "test"},
 			cols:  []string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX", "COLLATION", "NON_UNIQUE"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale", "extra"},
+			rows: [][]driver.Value{
+				{"a", "text", "text", "NO", nil, nil, nil, nil, nil},
+				{"b", "double", "double", "YES", nil, nil, 53, nil, nil},
+				{"c", "bigint", "bigint", "YES", nil, nil, 64, 0, nil}},
 		},
-		// Note: go-sqlmock mocks specify an ordered sequence
-		// of queries and results.  This (repeated) entry is
-		// needed because ProcessSQLData (redundantly) gets
-		// the set of tables via a SQL query.
 		{
-			query: "SELECT table_name FROM information_schema.tables where table_type = 'BASE TABLE' and (.+)",
-			args:  []driver.Value{"test"},
-			cols:  []string{"table_name"},
-			rows:  [][]driver.Value{{"test"}},
-		}, {
 			query: "SELECT (.+) FROM `test`.`test`",
 			cols:  []string{"a", "b", "c"},
 			rows: [][]driver.Value{
@@ -387,7 +373,7 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{"test"})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{"test", db})
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"test": ddl.CreateTable{
@@ -411,7 +397,7 @@ func TestProcessSQLData_MultiCol(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{"test"})
+	common.ProcessData(conv, InfoSchemaImpl{"test", db})
 	assert.Equal(t, []spannerData{
 		{table: "test", cols: []string{"a", "b", "synth_id"}, vals: []interface{}{"cat", float64(42.3), int64(0)}},
 		{table: "test", cols: []string{"a", "c", "synth_id"}, vals: []interface{}{"dog", int64(22), int64(-9223372036854775808)}}},
@@ -439,7 +425,7 @@ func TestSetRowStats(t *testing.T) {
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
 	conv.SetDataMode()
-	common.SetRowStats(conv, db, InfoSchemaImpl{"test"})
+	common.SetRowStats(conv, InfoSchemaImpl{"test", db})
 	assert.Equal(t, int64(5), conv.Stats.Rows["test1"])
 	assert.Equal(t, int64(142), conv.Stats.Rows["test2"])
 	assert.Equal(t, int64(0), conv.Unexpecteds())
diff --git a/sources/postgres/data_test.go b/sources/postgres/data_test.go
index aa05951..6c05ce0 100644
--- a/sources/postgres/data_test.go
+++ b/sources/postgres/data_test.go
@@ -280,12 +280,6 @@ func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *internal.Conv {
 	conv := internal.MakeConv()
 	conv.SpSchema[spTable.Name] = spTable
 	conv.SrcSchema[srcTable.Name] = srcTable
-	conv.ToSource[spTable.Name] = internal.NameAndCols{Name: srcTable.Name, Cols: make(map[string]string)}
-	conv.ToSpanner[srcTable.Name] = internal.NameAndCols{Name: spTable.Name, Cols: make(map[string]string)}
-	for i := range spTable.ColNames {
-		conv.ToSource[spTable.Name].Cols[spTable.ColNames[i]] = srcTable.ColNames[i]
-		conv.ToSpanner[srcTable.Name].Cols[srcTable.ColNames[i]] = spTable.ColNames[i]
-	}
 	return conv
 }
 
diff --git a/sources/postgres/infoschema_test.go b/sources/postgres/infoschema_test.go
index b623e7a..fe61a5c 100644
--- a/sources/postgres/infoschema_test.go
+++ b/sources/postgres/infoschema_test.go
@@ -37,7 +37,7 @@ type mockSpec struct {
 	rows  [][]driver.Value // Set of rows returned.
 }
 
-func TestProcessInfoSchema(t *testing.T) {
+func TestProcessSchema(t *testing.T) {
 	ms := []mockSpec{
 		{
 			query: "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'",
@@ -48,15 +48,6 @@ func TestProcessInfoSchema(t *testing.T) {
 				{"public", "product"},
 				{"public", "test"},
 				{"public", "test_ref"}},
-		},
-		{
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "user"},
-			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
-			rows: [][]driver.Value{
-				{"user_id", "text", nil, "NO", nil, nil, nil, nil},
-				{"name", "text", nil, "NO", nil, nil, nil, nil},
-				{"ref", "bigint", nil, "YES", nil, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "user"},
@@ -77,12 +68,12 @@ func TestProcessInfoSchema(t *testing.T) {
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "cart"},
+			args:  []driver.Value{"public", "user"},
 			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
 			rows: [][]driver.Value{
-				{"productid", "text", nil, "NO", nil, nil, nil, nil},
-				{"userid", "text", nil, "NO", nil, nil, nil, nil},
-				{"quantity", "bigint", nil, "YES", nil, nil, 64, 0}},
+				{"user_id", "text", nil, "NO", nil, nil, nil, nil},
+				{"name", "text", nil, "NO", nil, nil, nil, nil},
+				{"ref", "bigint", nil, "YES", nil, nil, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "cart"},
@@ -109,11 +100,12 @@ func TestProcessInfoSchema(t *testing.T) {
 			},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "product"},
+			args:  []driver.Value{"public", "cart"},
 			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
 			rows: [][]driver.Value{
-				{"product_id", "text", nil, "NO", nil, nil, nil, nil},
-				{"product_name", "text", nil, "NO", nil, nil, nil, nil}},
+				{"productid", "text", nil, "NO", nil, nil, nil, nil},
+				{"userid", "text", nil, "NO", nil, nil, nil, nil},
+				{"quantity", "bigint", nil, "YES", nil, nil, 64, 0}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "product"},
@@ -128,6 +120,28 @@ func TestProcessInfoSchema(t *testing.T) {
 			query: "SELECT (.+) FROM pg_index (.+)",
 			args:  []driver.Value{"public", "product"},
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"public", "product"},
+			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"product_id", "text", nil, "NO", nil, nil, nil, nil},
+				{"product_name", "text", nil, "NO", nil, nil, nil, nil}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}},
+		}, {
+			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
+			rows: [][]driver.Value{{"public", "test_ref", "id", "ref_id", "fk_test4"},
+				{"public", "test_ref", "txt", "ref_txt", "fk_test4"}},
+		}, {
+			query: "SELECT (.+) FROM pg_index (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
 			args:  []driver.Value{"public", "test"},
@@ -156,18 +170,18 @@ func TestProcessInfoSchema(t *testing.T) {
 				{"vc6", "character varying", nil, "YES", nil, 6, nil, nil}},
 		}, {
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"public", "test"},
+			args:  []driver.Value{"public", "test_ref"},
 			cols:  []string{"column_name", "constraint_type"},
-			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}},
+			rows: [][]driver.Value{
+				{"ref_id", "PRIMARY KEY"},
+				{"ref_txt", "PRIMARY KEY"}},
 		}, {
 			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
-			args:  []driver.Value{"public", "test"},
+			args:  []driver.Value{"public", "test_ref"},
 			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
-			rows: [][]driver.Value{{"public", "test_ref", "id", "ref_id", "fk_test4"},
-				{"public", "test_ref", "txt", "ref_txt", "fk_test4"}},
 		}, {
 			query: "SELECT (.+) FROM pg_index (.+)",
-			args:  []driver.Value{"public", "test"},
+			args:  []driver.Value{"public", "test_ref"},
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		}, {
 			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
@@ -177,26 +191,11 @@ func TestProcessInfoSchema(t *testing.T) {
 				{"ref_id", "bigint", nil, "NO", nil, nil, 64, 0},
 				{"ref_txt", "text", nil, "NO", nil, nil, nil, nil},
 				{"abc", "text", nil, "NO", nil, nil, nil, nil}},
-		}, {
-			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
-			args:  []driver.Value{"public", "test_ref"},
-			cols:  []string{"column_name", "constraint_type"},
-			rows: [][]driver.Value{
-				{"ref_id", "PRIMARY KEY"},
-				{"ref_txt", "PRIMARY KEY"}},
-		}, {
-			query: "SELECT (.+) FROM PG_CLASS (.+) JOIN PG_NAMESPACE (.+) JOIN PG_CONSTRAINT (.+)",
-			args:  []driver.Value{"public", "test_ref"},
-			cols:  []string{"TABLE_SCHEMA", "TABLE_NAME", "COLUMN_NAME", "REF_COLUMN_NAME", "CONSTRAINT_NAME"},
-		}, {
-			query: "SELECT (.+) FROM pg_index (.+)",
-			args:  []driver.Value{"public", "test_ref"},
-			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
 		},
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{db})
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"user": ddl.CreateTable{
@@ -288,13 +287,9 @@ func TestProcessInfoSchema(t *testing.T) {
 // handling of bad rows and table and column renaming. The core data
 // conversion work of ProcessSqlData is done by ConvertData, which is
 // extensively is tested by TestConvertSqlRow.
-func TestProcessSqlData(t *testing.T) {
+func TestProcessData(t *testing.T) {
 	ms := []mockSpec{
 		{
-			query: "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'",
-			cols:  []string{"table_schema", "table_name"},
-			rows:  [][]driver.Value{{"public", "te st"}},
-		}, {
 			query: `SELECT [*] FROM "public"."te st"`, // query is a regexp!
 			cols:  []string{"a a", " b", " c "},
 			rows: [][]driver.Value{
@@ -315,6 +310,7 @@ func TestProcessSqlData(t *testing.T) {
 			}},
 		schema.Table{
 			Name:     "te st",
+			Schema:   "public",
 			ColNames: []string{"a_a", "_b", "_c_"},
 			ColDefs: map[string]schema.Column{
 				"a a": schema.Column{Name: "a a", Type: schema.Type{Name: "float4"}},
@@ -327,7 +323,7 @@ func TestProcessSqlData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{})
+	common.ProcessData(conv, InfoSchemaImpl{db})
 
 	assert.Equal(t,
 		[]spannerData{
@@ -423,15 +419,6 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 			cols:  []string{"table_schema", "table_name"},
 			rows:  [][]driver.Value{{"public", "test"}},
 		}, {
-			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
-			args:  []driver.Value{"public", "test"},
-			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
-			rows: [][]driver.Value{
-				{"a", "text", nil, "NO", nil, nil, nil, nil},
-				{"b", "double precision", nil, "YES", nil, nil, 53, nil},
-				{"c", "bigint", nil, "YES", nil, nil, 64, 0}},
-		},
-		{
 			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
 			args:  []driver.Value{"public", "test"},
 			cols:  []string{"column_name", "constraint_type"},
@@ -446,15 +433,14 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 			query: "SELECT (.+) FROM pg_index (.+)",
 			args:  []driver.Value{"public", "test"},
 			cols:  []string{"index_name", "column_name", "column_position", "is_unique", "order"},
-		},
-		// Note: go-sqlmock mocks specify an ordered sequence
-		// of queries and results.  This (repeated) entry is
-		// needed because ProcessSqlData (redundantly) gets
-		// the set of tables via a SQL query.
-		{
-			query: "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'",
-			cols:  []string{"table_schema", "table_name"},
-			rows:  [][]driver.Value{{"public", "test"}},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"public", "test"},
+			cols:  []string{"column_name", "data_type", "data_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"a", "text", nil, "NO", nil, nil, nil, nil},
+				{"b", "double precision", nil, "YES", nil, nil, 53, nil},
+				{"c", "bigint", nil, "YES", nil, nil, 64, 0}},
 		}, {
 			query: `SELECT [*] FROM "public"."test"`, // query is a regexp!
 			cols:  []string{"a", "b", "c"},
@@ -465,7 +451,7 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessInfoSchema(conv, db, InfoSchemaImpl{})
+	err := common.ProcessSchema(conv, InfoSchemaImpl{db})
 	assert.Nil(t, err)
 	conv.SetDataMode()
 	var rows []spannerData
@@ -473,7 +459,7 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessSQLData(conv, db, InfoSchemaImpl{})
+	common.ProcessData(conv, InfoSchemaImpl{db})
 	assert.Equal(t, []spannerData{
 		{table: "test", cols: []string{"a", "b", "synth_id"}, vals: []interface{}{"cat", float64(42.3), int64(0)}},
 		{table: "test", cols: []string{"a", "c", "synth_id"}, vals: []interface{}{"dog", int64(22), int64(-9223372036854775808)}}},
@@ -500,7 +486,7 @@ func TestSetRowStats(t *testing.T) {
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
 	conv.SetDataMode()
-	common.SetRowStats(conv, db, InfoSchemaImpl{})
+	common.SetRowStats(conv, InfoSchemaImpl{db})
 	assert.Equal(t, int64(5), conv.Stats.Rows["test1"])
 	assert.Equal(t, int64(142), conv.Stats.Rows["test2"])
 	assert.Equal(t, int64(0), conv.Unexpecteds())