Repository: cloudspannerecosystem/harbourbridge
Pull request #66 (2020-06-26 11:40:20)

Added support for MySQL Information Schema
Imported [GO-MySQL-Driver](https://github.com/go-sql-driver/mysql) to use [database/sql](https://golang.org/pkg/database/sql/) package.

Dharmil-searce: > I'm happy with the PR.
> Please merge when all comments are resolved.

@nevinheintze Only those with write access to the repo can merge PRs. I do not have that access so request you to merge the PR.

> Should we just do both postgres and mysql via string? Any thoughts on the additional processing/conversion cost?

We can discuss this on the group.

Patch:
diff --git a/main.go b/main.go
index 10fad92..1e6d4f4 100644
--- a/main.go
+++ b/main.go
@@ -270,7 +270,7 @@ func schemaFromSQL(driver string) (*internal.Conv, error) {
 		return nil, err
 	}
 	conv := internal.MakeConv()
-	err = postgres.ProcessInfoSchema(conv, sourceDB)
+	err = ProcessInfoSchema(driver, conv, sourceDB)
 	if err != nil {
 		return nil, err
 	}
@@ -291,8 +291,10 @@ func dataFromSQL(driver string, config spanner.BatchWriterConfig, client *sp.Cli
 	if err != nil {
 		return nil, err
 	}
-
-	postgres.SetRowStats(conv, sourceDB)
+	err = SetRowStats(driver, conv, sourceDB)
+	if err != nil {
+		return nil, err
+	}
 	totalRows := conv.Rows()
 	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
 	rows := int64(0)
@@ -311,7 +313,10 @@ func dataFromSQL(driver string, config spanner.BatchWriterConfig, client *sp.Cli
 		func(table string, cols []string, vals []interface{}) {
 			writer.AddRow(table, cols, vals)
 		})
-	postgres.ProcessSQLData(conv, sourceDB)
+	err = ProcessSQLData(driver, conv, sourceDB)
+	if err != nil {
+		return nil, err
+	}
 	writer.Flush()
 	return writer, nil
 }
@@ -766,3 +771,41 @@ func ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 		return fmt.Errorf("process dump for driver %s not supported", driver)
 	}
 }
+
+// ProcessInfoSchema invokes process infoschema function from a sql package based on driver selected.
+func ProcessInfoSchema(driver string, conv *internal.Conv, db *sql.DB) error {
+	switch driver {
+	case MYSQL:
+		return mysql.ProcessInfoSchema(conv, db)
+	case POSTGRES:
+		return postgres.ProcessInfoSchema(conv, db)
+	default:
+		return fmt.Errorf("schema conversion for driver %s not supported", driver)
+	}
+}
+
+// SetRowStats invokes SetRowStats function from a sql package based on driver selected.
+func SetRowStats(driver string, conv *internal.Conv, db *sql.DB) error {
+	switch driver {
+	case MYSQL:
+		mysql.SetRowStats(conv, db)
+	case POSTGRES:
+		postgres.SetRowStats(conv, db)
+	default:
+		return fmt.Errorf("Could get 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 {
+	switch driver {
+	case MYSQL:
+		mysql.ProcessSQLData(conv, db)
+	case POSTGRES:
+		postgres.ProcessSQLData(conv, db)
+	default:
+		return fmt.Errorf("Data conversion for driver %s is not supported", driver)
+	}
+	return nil
+}
diff --git a/mysql/infoschema.go b/mysql/infoschema.go
new file mode 100644
index 0000000..c8a0391
--- /dev/null
+++ b/mysql/infoschema.go
@@ -0,0 +1,364 @@
+// Copyright 2020 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 mysql
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	_ "github.com/go-sql-driver/mysql" // The driver should be used via the database/sql package.
+	_ "github.com/lib/pq"
+)
+
+// ProcessInfoSchema performs schema conversion for source database
+// 'db'. We assume that the source database supports information
+// schema tables. These 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) error {
+	tables, err := getTables(db)
+	if err != nil {
+		return err
+	}
+	for _, t := range tables {
+		if err := processTable(conv, db, t); err != nil {
+			return err
+		}
+	}
+	schemaToDDL(conv)
+	conv.AddPrimaryKeys()
+	return nil
+}
+
+// ProcessSQLData performs data conversion for source database
+// 'db'. For each table, we extract data using a "SELECT (colNamesList)" query,
+// convert the data to Spanner data (based on the source and Spanner
+// schemas), and write it to Spanner.  If we can't get/process data
+// 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) {
+	// TODO: refactor to use the set of tables computed by
+	// ProcessInfoSchema instead of computing them again.
+	tables, err := getTables(db)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
+		return
+	}
+	for _, t := range tables {
+		srcTable := buildTableName(t.schema, t.name)
+		srcSchema, ok := conv.SrcSchema[srcTable]
+		if !ok {
+			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
+			conv.Unexpected(fmt.Sprintf("Can't get schemas for table %s", srcTable))
+			continue
+		}
+		srcCols := srcSchema.ColNames
+		if len(srcCols) == 0 {
+			conv.Unexpected(fmt.Sprintf("Couldn't get source columns for table %s ", t.name))
+			continue
+		}
+		colNameList := buildColNameList(srcSchema, srcCols)
+		// 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 %s FROM %s.%s;`, colNameList, t.schema, t.name)
+		rows, err := db.Query(q)
+		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))
+			continue
+		}
+		spCols, err := internal.GetSpannerCols(conv, srcTable, srcCols)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Couldn't get spanner columns for table %s : err = %s", t.name, err))
+			continue
+		}
+		spSchema, ok := conv.SpSchema[spTable]
+		if !ok {
+			conv.Stats.BadRows[srcTable] += conv.Stats.Rows[srcTable]
+			conv.Unexpected(fmt.Sprintf("Can't get schemas for table %s", srcTable))
+			continue
+		}
+		v, scanArgs := buildVals(len(srcCols))
+		for rows.Next() {
+			// get RawBytes from data.
+			err = rows.Scan(scanArgs...)
+			if err != nil {
+				conv.Unexpected(fmt.Sprintf("Couldn't process sql data row: %s", err))
+				// Scan failed, so we don't have any data to add to bad rows.
+				conv.StatsAddBadRow(srcTable, conv.DataMode())
+				continue
+			}
+			values := valsToStrings(v)
+			ProcessDataRow(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, values)
+		}
+	}
+}
+
+// Building list of column names to support mysql spatial datatypes instead of
+// using 'SELECT *' because spatial columns will be fetched using ST_AsText(colName).
+func buildColNameList(srcSchema schema.Table, srcColName []string) string {
+	var srcColTypes []string
+	var colList, colTmpName string
+	for _, colName := range srcColName {
+		// To handle cases where column name is reserved keyword or having space between words.
+		colTmpName = "`" + colName + "`"
+		srcColTypes = append(srcColTypes, srcSchema.ColDefs[colName].Type.Name)
+		for _, spatial := range MysqlSpatialDataTypes {
+			if strings.Contains(strings.ToLower(srcSchema.ColDefs[colName].Type.Name), spatial) {
+				colTmpName = "ST_AsText" + "(" + colTmpName + ")" + colTmpName
+				break
+			}
+		}
+		colList = colList + colTmpName + ","
+	}
+	return colList[:len(colList)-1]
+}
+
+// SetRowStats populates conv with the number of rows in each table.
+func SetRowStats(conv *internal.Conv, db *sql.DB) {
+	tables, err := getTables(db)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't get list of table: %s", err))
+		return
+	}
+	for _, t := range tables {
+		// 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;`, t.schema, t.name)
+		tableName := buildTableName(t.schema, t.name)
+		rows, err := db.Query(q)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Couldn't get number of rows for table %s", tableName))
+			continue
+		}
+		defer rows.Close()
+		var count int64
+		if rows.Next() {
+			err := rows.Scan(&count)
+			if err != nil {
+				conv.Unexpected(fmt.Sprintf("Can't get row count: %s", err))
+				continue
+			}
+			conv.Stats.Rows[tableName] += count
+		}
+	}
+}
+
+type schemaAndName struct {
+	schema string
+	name   string
+}
+
+func getTables(db *sql.DB) ([]schemaAndName, error) {
+	ignored := make(map[string]bool)
+	// Ignore all system tables: we just want to convert user tables.
+	for _, s := range []string{"information_schema", "mysql", "performance_schema", "sys"} {
+		ignored[s] = true
+	}
+	q := "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'"
+	rows, err := db.Query(q)
+	if err != nil {
+		return nil, fmt.Errorf("couldn't get tables: %w", err)
+	}
+	defer rows.Close()
+	var tableSchema, tableName string
+	var tables []schemaAndName
+	for rows.Next() {
+		rows.Scan(&tableSchema, &tableName)
+		if !ignored[tableSchema] {
+			tables = append(tables, schemaAndName{schema: tableSchema, name: tableName})
+		}
+	}
+	return tables, nil
+}
+
+func processTable(conv *internal.Conv, db *sql.DB, table schemaAndName) error {
+	cols, err := 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 := getConstraints(conv, db, table)
+	if err != nil {
+		return fmt.Errorf("couldn't get constraints for table %s.%s: %s", table.schema, table.name, err)
+	}
+	colDefs, colNames := processColumns(conv, cols, constraints)
+	name := buildTableName(table.schema, table.name)
+	var schemaPKeys []schema.Key
+	for _, k := range primaryKeys {
+		schemaPKeys = append(schemaPKeys, schema.Key{Column: k})
+	}
+	conv.SrcSchema[name] = schema.Table{
+		Name:        name,
+		ColNames:    colNames,
+		ColDefs:     colDefs,
+		PrimaryKeys: schemaPKeys}
+	return nil
+}
+
+func getColumns(table schemaAndName, db *sql.DB) (*sql.Rows, 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
+              FROM information_schema.COLUMNS c
+              where table_schema = ? and table_name = ? ORDER BY c.ordinal_position;`
+	return db.Query(q, table.schema, table.name)
+}
+
+func processColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string) {
+	colDefs := make(map[string]schema.Column)
+	var colNames []string
+	var colName, dataType, isNullable, columnType string
+	var colDefault sql.NullString
+	var charMaxLen, numericPrecision, numericScale sql.NullInt64
+	for cols.Next() {
+		err := cols.Scan(&colName, &dataType, &columnType, &isNullable, &colDefault, &charMaxLen, &numericPrecision, &numericScale)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		unique := false
+		ignored := schema.Ignored{}
+		for _, c := range constraints[colName] {
+			// c can be UNIQUE, PRIMARY KEY, FOREIGN KEY or CHECK
+			// We've already filtered out PRIMARY KEY.
+			switch c {
+			case "UNIQUE":
+				unique = true
+			case "FOREIGN KEY":
+				ignored.ForeignKey = true
+			case "CHECK":
+				ignored.Check = true
+			}
+		}
+		ignored.Default = colDefault.Valid
+		c := schema.Column{
+			Name:    colName,
+			Type:    toType(dataType, columnType, charMaxLen, numericPrecision, numericScale),
+			NotNull: toNotNull(conv, isNullable),
+			Unique:  unique,
+			Ignored: ignored,
+		}
+		colDefs[colName] = c
+		colNames = append(colNames, colName)
+	}
+	return colDefs, colNames
+}
+
+// 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.
+func getConstraints(conv *internal.Conv, db *sql.DB, table 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)
+	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(fmt.Sprintf("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
+}
+
+func toType(dataType string, columnType string, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
+	switch {
+	case dataType == "set":
+		return schema.Type{Name: dataType, ArrayBounds: []int64{-1}}
+	case charLen.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{charLen.Int64}}
+	case dataType == "decimal" && numericPrecision.Valid && numericScale.Valid && numericScale.Int64 != 0:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64, numericScale.Int64}}
+	case dataType == "decimal" && numericPrecision.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64}}
+	default:
+		return schema.Type{Name: dataType}
+	}
+}
+
+func toNotNull(conv *internal.Conv, isNullable string) bool {
+	switch isNullable {
+	case "YES":
+		return false
+	case "NO":
+		return true
+	}
+	conv.Unexpected(fmt.Sprintf("isNullable column has unknown value: %s", isNullable))
+	return false
+}
+
+// buildVals constructs []sql.RawBytes value containers to scan row
+// results into.  Returns both the underlying containers (as a slice)
+// as well as an interface{} of pointers to containers to pass to
+// rows.Scan.
+func buildVals(n int) (v []sql.RawBytes, iv []interface{}) {
+	v = make([]sql.RawBytes, n)
+	// rows.Scan wants '[]interface{}' as an argument, so we must copy the
+	// references into such a slice.
+	iv = make([]interface{}, len(v))
+	for i := range v {
+		iv[i] = &v[i]
+	}
+	return v, iv
+}
+
+func valsToStrings(vals []sql.RawBytes) []string {
+	toString := func(val sql.RawBytes) string {
+		if val == nil {
+			return "NULL"
+		}
+		return string(val)
+	}
+	var s []string
+	for _, v := range vals {
+		s = append(s, toString(v))
+	}
+	return s
+}
+
+func buildTableName(schema, name string) string {
+	return fmt.Sprintf("%s.%s", schema, name)
+}
diff --git a/mysql/toddl.go b/mysql/toddl.go
index 7424846..9608592 100644
--- a/mysql/toddl.go
+++ b/mysql/toddl.go
@@ -136,10 +136,10 @@ func toSpannerType(conv *internal.Conv, id string, mods []int64) (ddl.ScalarType
 		}
 		return ddl.String{Len: ddl.MaxLength{}}, nil
 	case "text", "tinytext", "mediumtext", "longtext":
-		maxExpectedMods(0)
+		maxExpectedMods(1)
 		return ddl.String{Len: ddl.MaxLength{}}, nil
 	case "set", "enum":
-		maxExpectedMods(0)
+		maxExpectedMods(1)
 		return ddl.String{Len: ddl.MaxLength{}}, nil
 	case "json":
 		maxExpectedMods(0)
@@ -148,7 +148,7 @@ func toSpannerType(conv *internal.Conv, id string, mods []int64) (ddl.ScalarType
 		maxExpectedMods(1)
 		return ddl.Bytes{Len: ddl.MaxLength{}}, nil
 	case "tinyblob", "mediumblob", "blob", "longblob":
-		maxExpectedMods(0)
+		maxExpectedMods(1)
 		return ddl.Bytes{Len: ddl.MaxLength{}}, nil
 	case "date":
 		maxExpectedMods(1)
diff --git a/postgres/infoschema.go b/postgres/infoschema.go
index d30b60d..444f617 100644
--- a/postgres/infoschema.go
+++ b/postgres/infoschema.go
@@ -190,7 +190,7 @@ func SetRowStats(conv *internal.Conv, db *sql.DB) {
 		if rows.Next() {
 			err := rows.Scan(&count)
 			if err != nil {
-				fmt.Printf("Can't get row count: %s\n", err)
+				conv.Unexpected(fmt.Sprintf("Can't get row count: %s", err))
 				continue
 			}
 			conv.Stats.Rows[tableName] += count
@@ -268,7 +268,7 @@ func processColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string]
 	for cols.Next() {
 		err := cols.Scan(&colName, &dataType, &elementDataType, &isNullable, &colDefault, &charMaxLen, &numericPrecision, &numericScale)
 		if err != nil {
-			fmt.Printf("Can't scan: %v\n", err)
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
 			continue
 		}
 		unique := false
@@ -320,11 +320,11 @@ func getConstraints(conv *internal.Conv, db *sql.DB, table schemaAndName) ([]str
 	for rows.Next() {
 		err := rows.Scan(&col, &constraint)
 		if err != nil {
-			fmt.Printf("Can't scan: %v\n", err)
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
 			continue
 		}
 		if col == "" || constraint == "" {
-			fmt.Printf("Got empty col or constraint\n")
+			conv.Unexpected(fmt.Sprintf("Got empty col or constraint"))
 			continue
 		}
 		switch constraint {


Test patch:
diff --git a/mysql/infoschema_test.go b/mysql/infoschema_test.go
new file mode 100644
index 0000000..51f10cc
--- /dev/null
+++ b/mysql/infoschema_test.go
@@ -0,0 +1,309 @@
+// Copyright 2020 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 mysql
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"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 TestProcessInfoSchemaMYSQL(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{
+				{"test", "cart"},
+				{"test", "test"}},
+		}, {
+			query: "SELECT (.+) FROM information_schema.COLUMNS (.+)",
+			args:  []driver.Value{"test", "cart"},
+			cols:  []string{"column_name", "data_type", "column_type", "is_nullable", "column_default", "character_maximum_length", "numeric_precision", "numeric_scale"},
+			rows: [][]driver.Value{
+				{"productid", "text", "text", "NO", nil, nil, nil, nil},
+				{"userid", "text", "text", "NO", nil, nil, nil, nil},
+				{"quantity", "bigint", "bigint", "YES", nil, nil, 64, 0}},
+		}, {
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"test", "cart"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows: [][]driver.Value{
+				{"productid", "PRIMARY KEY"},
+				{"userid", "PRIMARY KEY"}},
+		}, {
+			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"},
+			rows: [][]driver.Value{
+				{"id", "bigint", "bigint", "NO", nil, nil, 64, 0},
+				{"s", "set", "set", "YES", nil, nil, nil, nil},
+				{"txt", "text", "text", "NO", nil, nil, nil, nil},
+				{"b", "boolean", "boolean", "YES", nil, nil, nil, nil},
+				{"bs", "bigint", "bigint", "NO", "nextval('test11_bs_seq'::regclass)", nil, 64, 0},
+				{"bl", "blob", "blob", "YES", nil, nil, nil, nil},
+				{"c", "char", "char(1)", "YES", nil, 1, nil, nil},
+				{"c8", "char", "char(8)", "YES", nil, 8, nil, nil},
+				{"d", "date", "date", "YES", nil, nil, nil, nil},
+				{"f8", "double", "double", "YES", nil, nil, 53, nil},
+				{"f4", "float", "float", "YES", nil, nil, 24, nil},
+				{"i8", "bigint", "bigint", "YES", nil, nil, 64, 0},
+				{"i4", "integer", "integer", "YES", nil, nil, 32, 0},
+				{"i2", "smallint", "smallint", "YES", nil, nil, 16, 0},
+				{"si", "integer", "integer", "NO", "nextval('test11_s_seq'::regclass)", nil, 32, 0},
+				{"ts", "datetime", "datetime", "YES", nil, nil, nil, nil},
+				{"tz", "timestamp", "timestamp", "YES", nil, nil, nil, nil},
+				{"vc", "varchar", "varchar", "YES", nil, nil, nil, nil},
+				{"vc6", "varchar", "varchar(6)", "YES", nil, 6, 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"}},
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := internal.MakeConv()
+	err := ProcessInfoSchema(conv, db)
+	assert.Nil(t, err)
+	expectedSchema := map[string]ddl.CreateTable{
+		"test_cart": ddl.CreateTable{
+			Name:     "test_cart",
+			ColNames: []string{"productid", "userid", "quantity"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"productid": ddl.ColumnDef{Name: "productid", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true},
+				"userid":    ddl.ColumnDef{Name: "userid", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true},
+				"quantity":  ddl.ColumnDef{Name: "quantity", T: ddl.Int64{}},
+			},
+			Pks: []ddl.IndexKey{ddl.IndexKey{Col: "productid"}, ddl.IndexKey{Col: "userid"}}},
+		"test_test": ddl.CreateTable{
+			Name:     "test_test",
+			ColNames: []string{"id", "s", "txt", "b", "bs", "bl", "c", "c8", "d", "f8", "f4", "i8", "i4", "i2", "si", "ts", "tz", "vc", "vc6"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"id":  ddl.ColumnDef{Name: "id", T: ddl.Int64{}, NotNull: true},
+				"s":   ddl.ColumnDef{Name: "s", T: ddl.String{Len: ddl.MaxLength{}}, IsArray: true},
+				"txt": ddl.ColumnDef{Name: "txt", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true},
+				"b":   ddl.ColumnDef{Name: "b", T: ddl.Bool{}},
+				"bs":  ddl.ColumnDef{Name: "bs", T: ddl.Int64{}, NotNull: true},
+				"bl":  ddl.ColumnDef{Name: "bl", T: ddl.Bytes{Len: ddl.MaxLength{}}},
+				"c":   ddl.ColumnDef{Name: "c", T: ddl.String{Len: ddl.Int64Length{Value: 1}}},
+				"c8":  ddl.ColumnDef{Name: "c8", T: ddl.String{Len: ddl.Int64Length{Value: 8}}},
+				"d":   ddl.ColumnDef{Name: "d", T: ddl.Date{}},
+				"f8":  ddl.ColumnDef{Name: "f8", T: ddl.Float64{}},
+				"f4":  ddl.ColumnDef{Name: "f4", T: ddl.Float64{}},
+				"i8":  ddl.ColumnDef{Name: "i8", T: ddl.Int64{}},
+				"i4":  ddl.ColumnDef{Name: "i4", T: ddl.Int64{}},
+				"i2":  ddl.ColumnDef{Name: "i2", T: ddl.Int64{}},
+				"si":  ddl.ColumnDef{Name: "si", T: ddl.Int64{}, NotNull: true},
+				"ts":  ddl.ColumnDef{Name: "ts", T: ddl.Timestamp{}},
+				"tz":  ddl.ColumnDef{Name: "tz", T: ddl.Timestamp{}},
+				"vc":  ddl.ColumnDef{Name: "vc", T: ddl.String{Len: ddl.MaxLength{}}},
+				"vc6": ddl.ColumnDef{Name: "vc6", T: ddl.String{Len: ddl.Int64Length{Value: 6}}},
+			},
+			Pks: []ddl.IndexKey{ddl.IndexKey{Col: "id"}}},
+	}
+	assert.Equal(t, expectedSchema, stripSchemaComments(conv.SpSchema))
+	assert.Equal(t, len(conv.Issues["test.cart"]), 0)
+	expectedIssues := map[string][]internal.SchemaIssue{
+		"bs": []internal.SchemaIssue{internal.DefaultValue},
+		"f4": []internal.SchemaIssue{internal.Widened},
+		"i4": []internal.SchemaIssue{internal.Widened},
+		"i2": []internal.SchemaIssue{internal.Widened},
+		"si": []internal.SchemaIssue{internal.Widened, internal.DefaultValue},
+		"ts": []internal.SchemaIssue{internal.Datetime},
+	}
+	assert.Equal(t, expectedIssues, conv.Issues["test.test"])
+	assert.Equal(t, int64(0), conv.Unexpecteds())
+}
+
+func TestProcessSQLData(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{{"test", "te st"}},
+		}, {
+			query: "SELECT (.+) FROM test.te st",
+			cols:  []string{"a a", " b", " c "},
+			rows: [][]driver.Value{
+				{42.3, 3, "cat"},
+				{6.6, 22, "dog"},
+				{6.6, "2006-01-02", "dog"}}, // Test bad row logic.
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := buildConv(
+		ddl.CreateTable{
+			Name:     "test_te_st",
+			ColNames: []string{"a a", " b", " c "},
+			ColDefs: map[string]ddl.ColumnDef{
+				"a_a": ddl.ColumnDef{Name: "a_a", T: ddl.Float64{}},
+				"Ab":  ddl.ColumnDef{Name: "Ab", T: ddl.Int64{}},
+				"Ac_": ddl.ColumnDef{Name: "Ac_", T: ddl.String{Len: ddl.MaxLength{}}},
+			}},
+		schema.Table{
+			Name:     "test.te st",
+			ColNames: []string{"a_a", "_b", "_c_"},
+			ColDefs: map[string]schema.Column{
+				"a a": schema.Column{Name: "a a", Type: schema.Type{Name: "float"}},
+				" b":  schema.Column{Name: " b", Type: schema.Type{Name: "int"}},
+				" c ": schema.Column{Name: " c ", Type: schema.Type{Name: "text"}},
+			}})
+
+	conv.SetDataMode()
+	var rows []spannerData
+	conv.SetDataSink(
+		func(table string, cols []string, vals []interface{}) {
+			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
+		})
+	ProcessSQLData(conv, db)
+	assert.Equal(t,
+		[]spannerData{
+			spannerData{table: "test_te_st", cols: []string{"a_a", "Ab", "Ac_"}, vals: []interface{}{float64(42.3), int64(3), "cat"}},
+			spannerData{table: "test_te_st", cols: []string{"a_a", "Ab", "Ac_"}, vals: []interface{}{float64(6.6), int64(22), "dog"}},
+		},
+		rows)
+	assert.Equal(t, conv.BadRows(), int64(1))
+	assert.Equal(t, conv.SampleBadRows(10), []string{"table=test.te st cols=[a a  b  c ] data=[6.6 2006-01-02 dog]\n"})
+	assert.Equal(t, int64(1), conv.Unexpecteds()) // Bad row generates an entry in unexpected.
+}
+
+func TestProcessSQLData_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
+	// i.e. ProcessSQLData uses the schemas built by
+	// ProcessInfoSchema.
+	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{{"test", "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"},
+			rows: [][]driver.Value{
+				{"a", "text", "text", "NO", nil, nil, nil, nil},
+				{"b", "double", "double", "YES", nil, nil, 53, nil},
+				{"c", "bigint", "bigint", "YES", nil, nil, 64, 0}},
+		},
+		{
+			query: "SELECT (.+) FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS (.+)",
+			args:  []driver.Value{"test", "test"},
+			cols:  []string{"column_name", "constraint_type"},
+			rows:  [][]driver.Value{},
+		},
+		{
+			query: "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'",
+			cols:  []string{"table_schema", "table_name"},
+			rows:  [][]driver.Value{{"test", "test"}},
+		}, {
+			query: "SELECT (.+) FROM test.test",
+			cols:  []string{"a", "b", "c"},
+			rows: [][]driver.Value{
+				{"cat", 42.3, nil},
+				{"dog", nil, 22}},
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := internal.MakeConv()
+	err := ProcessInfoSchema(conv, db)
+	assert.Nil(t, err)
+	expectedSchema := map[string]ddl.CreateTable{
+		"test_test": ddl.CreateTable{
+			Name:     "test_test",
+			ColNames: []string{"a", "b", "c", "synth_id"},
+			ColDefs: map[string]ddl.ColumnDef{
+				"a":        ddl.ColumnDef{Name: "a", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true},
+				"b":        ddl.ColumnDef{Name: "b", T: ddl.Float64{}},
+				"c":        ddl.ColumnDef{Name: "c", T: ddl.Int64{}},
+				"synth_id": ddl.ColumnDef{Name: "synth_id", T: ddl.Int64{}},
+			},
+			Pks: []ddl.IndexKey{ddl.IndexKey{Col: "synth_id"}}},
+	}
+	assert.Equal(t, expectedSchema, stripSchemaComments(conv.SpSchema))
+	expectedIssues := map[string][]internal.SchemaIssue{}
+	assert.Equal(t, expectedIssues, conv.Issues["test.test"])
+	assert.Equal(t, int64(0), conv.Unexpecteds())
+	conv.SetDataMode()
+	var rows []spannerData
+	conv.SetDataSink(
+		func(table string, cols []string, vals []interface{}) {
+			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
+		})
+	ProcessSQLData(conv, db)
+	assert.Equal(t, []spannerData{
+		{table: "test_test", cols: []string{"a", "b", "synth_id"}, vals: []interface{}{"cat", float64(42.3), int64(0)}},
+		{table: "test_test", cols: []string{"a", "c", "synth_id"}, vals: []interface{}{"dog", int64(22), int64(-9223372036854775808)}}},
+		rows)
+	assert.Equal(t, int64(0), conv.Unexpecteds())
+}
+
+func TestSetRowStats(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{{"test", "test1"}, {"test", "test2"}},
+		}, {
+			query: `SELECT COUNT[(][*][)] FROM test.test1`,
+			cols:  []string{"count"},
+			rows:  [][]driver.Value{{5}},
+		}, {
+			query: `SELECT COUNT[(][*][)] FROM test.test2`,
+			cols:  []string{"count"},
+			rows:  [][]driver.Value{{142}},
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := internal.MakeConv()
+	conv.SetDataMode()
+	SetRowStats(conv, db)
+	assert.Equal(t, int64(5), conv.Stats.Rows["test.test1"])
+	assert.Equal(t, int64(142), conv.Stats.Rows["test.test2"])
+	assert.Equal(t, int64(0), conv.Unexpecteds())
+}
+
+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
+}