Repository: cloudspannerecosystem/harbourbridge
Pull request #58 (2020-04-20 07:00:14)

Complete information-schema mode: add data processing.
In information-schema mode we use SQL queries to get schema
information and data from a source database. A previous commit had
added schema processing (reading information_schema tables). This
commit adds the data processing part (reading data from tables using
"SELECT *" queries).


Patch:
diff --git a/go.sum b/go.sum
index 712a643..8cfcffc 100644
--- a/go.sum
+++ b/go.sum
@@ -265,6 +265,7 @@ golang.org/x/tools v0.0.0-20200317043434-63da46f3035e/go.mod h1:Sl4aGygMT6LrqrWc
 golang.org/x/tools v0.0.0-20200318054722-11a475a590ac/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
diff --git a/internal/convert.go b/internal/convert.go
index 28d3281..687d611 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -179,6 +179,19 @@ func (conv *Conv) GetDDL(c ddl.Config) []string {
 	return ddl
 }
 
+// WriteRow calls dataSink and updates row stats.
+func (conv *Conv) WriteRow(srcTable, spTable string, spCols []string, spVals []interface{}) {
+	if conv.dataSink == nil {
+		msg := "Internal error: ProcessDataRow called but dataSink not configured"
+		VerbosePrintf("%s\n", msg)
+		conv.unexpected(msg)
+		conv.statsAddBadRow(srcTable, conv.dataMode())
+	} else {
+		conv.dataSink(spTable, spCols, spVals)
+		conv.statsAddGoodRow(srcTable, conv.dataMode())
+	}
+}
+
 // Rows returns the total count of data rows processed.
 func (conv *Conv) Rows() int64 {
 	n := int64(0)
@@ -222,6 +235,18 @@ func (conv *Conv) Unexpecteds() int64 {
 	return int64(len(conv.stats.unexpected))
 }
 
+// CollectBadRows updates the list of bad rows, while respecting
+// the byte limit for bad rows.
+func (conv *Conv) CollectBadRow(srcTable string, srcCols, vals []string) {
+	r := &row{table: srcTable, cols: srcCols, vals: vals}
+	bytes := byteSize(r)
+	// Cap storage used by badRows. Keep at least one bad row.
+	if len(conv.sampleBadRows.rows) == 0 || bytes+conv.sampleBadRows.bytes < conv.sampleBadRows.bytesLimit {
+		conv.sampleBadRows.rows = append(conv.sampleBadRows.rows, r)
+		conv.sampleBadRows.bytes += bytes
+	}
+}
+
 // SampleBadRows returns a string-formatted list of rows that generated errors.
 // Returns at most n rows.
 func (conv *Conv) SampleBadRows(n int) []string {
@@ -300,6 +325,10 @@ func (conv *Conv) statsAddRow(srcTable string, b bool) {
 	}
 }
 
+func (conv *Conv) statsAddRows(srcTable string, count int64) {
+	conv.stats.rows[srcTable] += count
+}
+
 // statsAddGoodRow increments the good-row stats for 'srcTable' if b
 // is true.  See statsAddRow comments for context.
 func (conv *Conv) statsAddGoodRow(srcTable string, b bool) {
@@ -308,6 +337,10 @@ func (conv *Conv) statsAddGoodRow(srcTable string, b bool) {
 	}
 }
 
+func (conv *Conv) statsAddGoodRows(srcTable string, count int64) {
+	conv.stats.goodRows[srcTable] += count
+}
+
 // statsAddBadRow increments the bad-row stats for 'srcTable' if b is
 // true.  See statsAddRow comments for context.
 func (conv *Conv) statsAddBadRow(srcTable string, b bool) {
@@ -316,6 +349,10 @@ func (conv *Conv) statsAddBadRow(srcTable string, b bool) {
 	}
 }
 
+func (conv *Conv) statsAddBadRows(srcTable string, count int64) {
+	conv.stats.badRows[srcTable] += count
+}
+
 func prNodeType(n nodes.Node) string {
 	// Strip off "pg_query." prefix from nodes.Nodes type.
 	return strings.TrimPrefix(reflect.TypeOf(n).Name(), "pg_query.")
diff --git a/internal/data.go b/internal/data.go
index 3cd229d..d2ce0f0 100644
--- a/internal/data.go
+++ b/internal/data.go
@@ -38,23 +38,9 @@ func ProcessDataRow(conv *Conv, srcTable string, srcCols, vals []string) {
 	if err != nil {
 		conv.unexpected(fmt.Sprintf("Error while converting data: %s\n", err))
 		conv.statsAddBadRow(srcTable, conv.dataMode())
-		r := &row{table: srcTable, cols: srcCols, vals: vals}
-		bytes := byteSize(r)
-		// Cap storage used by badRows. Keep at least one bad row.
-		if len(conv.sampleBadRows.rows) == 0 || bytes+conv.sampleBadRows.bytes < conv.sampleBadRows.bytesLimit {
-			conv.sampleBadRows.rows = append(conv.sampleBadRows.rows, r)
-			conv.sampleBadRows.bytes += bytes
-		}
+		conv.CollectBadRow(srcTable, srcCols, vals)
 	} else {
-		if conv.dataSink == nil {
-			msg := "Internal error: ProcessDataRow called but dataSink not configured"
-			VerbosePrintf("%s\n", msg)
-			conv.unexpected(msg)
-			conv.statsAddBadRow(srcTable, conv.dataMode())
-		} else {
-			conv.dataSink(spTable, spCols, spVals)
-			conv.statsAddGoodRow(srcTable, conv.dataMode())
-		}
+		conv.WriteRow(srcTable, spTable, spCols, spVals)
 	}
 }
 
@@ -203,7 +189,7 @@ func convTimestamp(srcTypeName string, location *time.Location, val string) (t t
 	// pg_dump outputs timestamps as ISO 8601, except:
 	// a) it uses space instead of T
 	// b) timezones are abbreviated to just hour (minute is specified only if non-zero).
-	if srcTypeName == "timestamptz" {
+	if srcTypeName == "timestamptz" || srcTypeName == "timestamp with time zone" {
 		// PostgreSQL abbreviates timezone to just hour where possible.
 		t, err = time.Parse("2006-01-02 15:04:05Z07", val)
 		if err != nil {
diff --git a/internal/infoschema.go b/internal/infoschema.go
index 25fa371..36558bb 100644
--- a/internal/infoschema.go
+++ b/internal/infoschema.go
@@ -17,23 +17,28 @@ package internal
 import (
 	"database/sql"
 	"fmt"
+	"math/bits"
+	"reflect"
+	"strconv"
+	"time"
 
+	"cloud.google.com/go/civil"
 	_ "github.com/lib/pq"
 
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
+// TODO: All of the queries to get tables and table data should be in
+// a single transaction to ensure we obtain a consistent snapshot of
+// schema information and table data (pg_dump does something
+// similar).
+
 // 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 *Conv, db *sql.DB) error {
-	// TODO: All of the queries to get tables and table data
-	// should be in a single transaction to ensure we obtain a
-	// consistent snapshot of schema information across tables
-	// (pg_dump does something similar). When we add SELECT
-	// queries to get data, we should wrap those in the same
-	// transaction to ensure consistency of schema and data.
 	tables, err := getTables(db)
 	if err != nil {
 		return err
@@ -48,6 +53,150 @@ func ProcessInfoSchema(conv *Conv, db *sql.DB) error {
 	return nil
 }
 
+// ProcessSqlData performs data conversion for source database
+// 'db'. For each table, we extract data using a "SELECT *" query,
+// convert the data to Spanner data (based on the source and Spanner
+// schemas), and write it to Spanner.  If we can't get/process data
+// for a table, we skip that table and process the remaining tables.
+//
+// Note that the database/sql library has a somewhat complex model for
+// returning data from rows.Scan. Scalar values can be returned using
+// the native value used by the underlying driver (by passing
+// *interface{} to rows.Scan), or they can be converted to specific go
+// types. Array values are always returned as []byte, a string
+// encoding of the array values. This string encoding is
+// database/driver specific. For example, for PostgreSQL, array values
+// are returned in the form "{v1,v2,..,vn}", where each v1,v2,...,vn
+// is a PostgreSQL encoding of the respective array value.
+//
+// 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 ProcessSqlData(conv *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 {
+		// 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";`, t.schema, t.name)
+		rows, err := db.Query(q)
+		if err != nil {
+			conv.unexpected(fmt.Sprintf("Couldn't get data for table: %s", err))
+			continue
+		}
+		defer rows.Close()
+		srcTable := buildTableName(t.schema, t.name)
+		srcCols, err1 := rows.Columns()
+		spTable, err2 := GetSpannerTable(conv, srcTable)
+		spCols, err3 := GetSpannerCols(conv, srcTable, srcCols)
+		spSchema, ok1 := conv.spSchema[spTable]
+		srcSchema, ok2 := conv.srcSchema[srcTable]
+		if err1 != nil || err2 != nil || err3 != nil || !ok1 || !ok2 {
+			conv.statsAddBadRows(srcTable, conv.stats.rows[srcTable])
+			conv.unexpected(fmt.Sprintf("Can't get cols and schemas for table %s: err1=%s, err2=%s, err3=%s, ok1=%t, ok2=%t",
+				srcTable, err1, err2, err3, ok1, ok2))
+			continue
+		}
+		v, iv := buildVals(len(srcCols))
+		for rows.Next() {
+			err := rows.Scan(iv...)
+			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
+			}
+			cvtCols, cvtVals, err := ConvertSqlRow(conv, srcTable, srcCols, srcSchema, spTable, spCols, spSchema, v)
+			if err != nil {
+				conv.unexpected(fmt.Sprintf("Couldn't process sql data row: %s", err))
+				conv.statsAddBadRow(srcTable, conv.dataMode())
+				conv.CollectBadRow(srcTable, srcCols, valsToStrings(v))
+				continue
+			}
+			conv.WriteRow(srcTable, spTable, cvtCols, cvtVals)
+		}
+	}
+}
+
+// ConvertSqlRow performs data conversion for a single row of data
+// returned from a 'SELECT *' query. ConvertSqlRow assumes that
+// srcCols, spCols and srcVals all have the same length. Note that
+// ConvertSqlRow returns cols as well as converted values. This is
+// because cols can change when we add a column (synthetic primary
+// key) or because we drop columns (handling of NULL values).
+func ConvertSqlRow(conv *Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, srcVals []interface{}) ([]string, []interface{}, error) {
+	var vs []interface{}
+	var cs []string
+	for i := range srcCols {
+		srcCd, ok1 := srcSchema.ColDefs[srcCols[i]]
+		spCd, ok2 := spSchema.ColDefs[spCols[i]]
+		if !ok1 || !ok2 {
+			return nil, nil, fmt.Errorf("data conversion: can't find schema for column %s of table %s", srcCols[i], srcTable)
+		}
+		if srcVals[i] == nil {
+			continue // Skip NULL values (nil is used by database/sql to represent NULL values).
+		}
+		var spVal interface{}
+		var err error
+		if spCd.IsArray {
+			spVal, err = cvtSqlArray(conv, srcCd, spCd, srcVals[i])
+		} else {
+			spVal, err = cvtSqlScalar(conv, srcCd, spCd, srcVals[i])
+		}
+		if err != nil { // Skip entire row if we hit error.
+			return nil, nil, fmt.Errorf("can't convert sql data for column %s of table %s: %w", srcCols[i], srcTable, err)
+		}
+		vs = append(vs, spVal)
+		cs = append(cs, srcCols[i])
+	}
+	if aux, ok := conv.syntheticPKeys[spTable]; ok {
+		cs = append(cs, aux.col)
+		vs = append(vs, int64(bits.Reverse64(uint64(aux.sequence))))
+		aux.sequence++
+		conv.syntheticPKeys[spTable] = aux
+	}
+	return cs, vs, nil
+}
+
+// SetRowStats populates conv with the number of rows in each table.
+func SetRowStats(conv *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 {
+		// 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";`, 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 {
+				fmt.Printf("Can't get row count: %s\n", err)
+				continue
+			}
+			conv.statsAddRows(tableName, count)
+		}
+	}
+}
+
 type schemaAndName struct {
 	schema string // PostgreSQL schema (aka namespace for PostgreSQL objects).
 	name   string
@@ -62,7 +211,7 @@ func getTables(db *sql.DB) ([]schemaAndName, error) {
 	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\n", err)
+		return nil, fmt.Errorf("couldn't get tables: %w\n", err)
 	}
 	defer rows.Close()
 	var tableSchema, tableName string
@@ -79,18 +228,15 @@ func getTables(db *sql.DB) ([]schemaAndName, error) {
 func processTable(conv *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\n", table.schema, table.name, err)
+		return fmt.Errorf("couldn't get schema for table %s.%s: %s\n", 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\n", table.schema, table.name, err)
+		return fmt.Errorf("couldn't get constraints for table %s.%s: %s\n", table.schema, table.name, err)
 	}
 	colDefs, colNames := processColumns(conv, cols, constraints)
-	name := fmt.Sprintf("%s.%s", table.schema, table.name)
-	if table.schema == "public" { // Drop 'public' prefix.
-		name = table.name
-	}
+	name := buildTableName(table.schema, table.name)
 	var schemaPKeys []schema.Key
 	for _, k := range primaryKeys {
 		schemaPKeys = append(schemaPKeys, schema.Key{Column: k})
@@ -217,3 +363,137 @@ func toNotNull(conv *Conv, isNullable string) bool {
 	conv.unexpected(fmt.Sprintf("isNullable column has unknown value: %s", isNullable))
 	return false
 }
+
+func cvtSqlArray(conv *Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interface{}) (interface{}, error) {
+	a, ok := val.([]byte)
+	if !ok {
+		return nil, fmt.Errorf("can't convert array values to []byte")
+	}
+	return convArray(spCd.T, srcCd.Type.Name, conv.location, string(a))
+}
+
+// cvtSqlScalar converts a values returned from a SQL query to a
+// Spanner value.  In principle, we could just hand the values we get
+// from the driver over to Spanner and have the Spanner client handle
+// conversions and errors. However we handle the conversions
+// explicitly ourselves so that we can generate more targeted error
+// messages. Note that the caller is responsible for handling nil
+// values (used to represent NULL). We handle each of the remaining
+// cases of values returned by the database/sql library:
+//    bool
+//    []byte
+//    int64
+//    float64
+//    string
+//    time.Time
+func cvtSqlScalar(conv *Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interface{}) (interface{}, error) {
+	switch spCd.T.(type) {
+	case ddl.Bool:
+		switch v := val.(type) {
+		case bool:
+			return v, nil
+		case string:
+			return convBool(v)
+		}
+	case ddl.Bytes:
+		switch v := val.(type) {
+		case []byte:
+			return v, nil
+		}
+	case ddl.Date:
+		// The PostgreSQL driver uses time.Time to represent
+		// dates.  Note that the database/sql library doesn't
+		// document how dates are represented, so maybe this
+		// isn't a driver issue, but a generic database/sql
+		// issue.  We explicitly convert from time.Time to
+		// civil.Date (used by the Spanner client library).
+		switch v := val.(type) {
+		case string:
+			return convDate(v)
+		case time.Time:
+			return civil.DateOf(v), nil
+		}
+	case ddl.Int64:
+		switch v := val.(type) {
+		case []byte: // Parse as int64.
+			return convInt64(string(v))
+		case int64:
+			return v, nil
+		case float64: // Truncate.
+			return int64(v), nil
+		case string: // Parse as int64.
+			return convInt64(v)
+		}
+	case ddl.Float64:
+		switch v := val.(type) {
+		case []byte: // Note: PostgreSQL uses []byte for numeric.
+			return convFloat64(string(v))
+		case int64:
+			return float64(v), nil
+		case float64:
+			return v, nil
+		case string:
+			return convFloat64(v)
+		}
+	case ddl.String:
+		switch v := val.(type) {
+		case bool:
+			return strconv.FormatBool(v), nil
+		case []byte:
+			return string(v), nil
+		case int64:
+			return strconv.FormatInt(v, 10), nil
+		case float64:
+			return strconv.FormatFloat(v, 'g', -1, 64), nil
+		case string:
+			return v, nil
+		case time.Time:
+			return v.String(), nil
+		}
+	case ddl.Timestamp:
+		switch v := val.(type) {
+		case string:
+			return convTimestamp(srcCd.Type.Name, conv.location, v)
+		case time.Time:
+			return v, nil
+		}
+	}
+	return nil, fmt.Errorf("can't convert value of type %s to Spanner type %s", reflect.TypeOf(val), reflect.TypeOf(spCd.T))
+}
+
+// buildVals contructs interface{} 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 []interface{}, iv []interface{}) {
+	v = make([]interface{}, n)
+	for i := range v {
+		iv = append(iv, &v[i])
+	}
+	return v, iv
+}
+
+func valsToStrings(vals []interface{}) []string {
+	toString := func(val interface{}) string {
+		if val == nil {
+			return "NULL"
+		}
+		switch v := val.(type) {
+		case *interface{}:
+			val = *v
+		}
+		return fmt.Sprintf("%v", val)
+	}
+	var s []string
+	for _, v := range vals {
+		s = append(s, toString(v))
+	}
+	return s
+}
+
+func buildTableName(schema, name string) string {
+	if schema == "public" { // Drop 'public' prefix.
+		return name
+	}
+	return fmt.Sprintf("%s.%s", schema, name)
+}
diff --git a/main.go b/main.go
index ba37077..43c4c38 100644
--- a/main.go
+++ b/main.go
@@ -203,7 +203,7 @@ func dataConv(driver string, ioHelper *ioStreams, client *sp.Client, conv *inter
 	}
 	switch driver {
 	case POSTGRES:
-		return dataFromSQL(config, client, conv)
+		return dataFromSQL(config, client, conv, driver)
 	case PGDUMP:
 		return dataFromPgDump(config, ioHelper, client, conv)
 	default:
@@ -234,12 +234,6 @@ func pgDriverConfig() (string, error) {
 }
 
 func schemaFromSQL(driver string) (*internal.Conv, error) {
-	fmt.Println(`###################################################################
-	Accessing a source DB via an database/sql driver is an experimental
-	feature that is not fully implemented. Currently we only do schema
-	conversion and we only support PostgreSQL.
-	###################################################################`)
-
 	driverConfig, err := driverConfig(driver)
 	if err != nil {
 		return nil, err
@@ -256,8 +250,43 @@ func schemaFromSQL(driver string) (*internal.Conv, error) {
 	return conv, nil
 }
 
-func dataFromSQL(config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
-	return spanner.NewBatchWriter(config), nil
+func dataFromSQL(config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv, driver string) (*spanner.BatchWriter, error) {
+	// TODO: Refactor to avoid redundant calls to driverConfig and
+	// Open in schemaFromSQL and dataFromSQL. Also refactor to
+	// share code with dataFromPgDump. Use single transaction for
+	// reading schema and data from source db to get consistent
+	// dump.
+	driverConfig, err := driverConfig(driver)
+	if err != nil {
+		return nil, err
+	}
+	sourceDB, err := sql.Open(driver, driverConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	internal.SetRowStats(conv, sourceDB)
+	totalRows := conv.Rows()
+	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
+	rows := int64(0)
+	config.Write = func(m []*sp.Mutation) error {
+		_, err := client.Apply(context.Background(), m)
+		if err != nil {
+			return err
+		}
+		atomic.AddInt64(&rows, int64(len(m)))
+		p.MaybeReport(atomic.LoadInt64(&rows))
+		return nil
+	}
+	writer := spanner.NewBatchWriter(config)
+	conv.SetDataMode()
+	conv.SetDataSink(
+		func(table string, cols []string, vals []interface{}) {
+			writer.AddRow(table, cols, vals)
+		})
+	internal.ProcessSqlData(conv, sourceDB)
+	writer.Flush()
+	return writer, nil
 }
 
 type ioStreams struct {


Test patch:
diff --git a/internal/data_test.go b/internal/data_test.go
index eaba589..27e582d 100644
--- a/internal/data_test.go
+++ b/internal/data_test.go
@@ -283,9 +283,9 @@ func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *Conv {
 	conv.srcSchema[srcTable.Name] = srcTable
 	conv.toSource[spTable.Name] = nameAndCols{name: srcTable.Name, cols: make(map[string]string)}
 	conv.toSpanner[srcTable.Name] = nameAndCols{name: spTable.Name, cols: make(map[string]string)}
-	for _, c := range spTable.ColNames {
-		conv.toSource[spTable.Name].cols[c] = c
-		conv.toSpanner[srcTable.Name].cols[c] = c
+	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/internal/infoschema_test.go b/internal/infoschema_test.go
index 71359ca..168823b 100644
--- a/internal/infoschema_test.go
+++ b/internal/infoschema_test.go
@@ -15,25 +15,27 @@
 package internal
 
 import (
+	"database/sql"
 	"database/sql/driver"
 	"testing"
+	"time"
 
+	"cloud.google.com/go/spanner"
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
 )
 
-func TestProcessInfoSchema(t *testing.T) {
-	type mockSpec struct {
-		query string
-		args  []driver.Value   // Query args.
-		cols  []string         // Columns names for returned rows.
-		rows  [][]driver.Value // Set of rows returned.
-	}
-	db, mock, err := sqlmock.New()
-	assert.Nil(t, err)
+type mockSpec struct {
+	query string
+	args  []driver.Value   // Query args.
+	cols  []string         // Columns names for returned rows.
+	rows  [][]driver.Value // Set of rows returned.
+}
 
-	for _, m := range []mockSpec{
+func TestProcessInfoSchema(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"},
@@ -88,19 +90,10 @@ func TestProcessInfoSchema(t *testing.T) {
 			cols:  []string{"column_name", "constraint_type"},
 			rows:  [][]driver.Value{{"id", "PRIMARY KEY"}},
 		},
-	} {
-		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)
-		}
 	}
+	db := mkMockDB(t, ms)
 	conv := MakeConv()
-	err = ProcessInfoSchema(conv, db)
+	err := ProcessInfoSchema(conv, db)
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"cart": ddl.CreateTable{
@@ -153,4 +146,234 @@ func TestProcessInfoSchema(t *testing.T) {
 		"ts":   []schemaIssue{timestamp},
 	}
 	assert.Equal(t, expectedIssues, conv.issues["test"])
+	assert.Equal(t, int64(0), conv.Unexpecteds())
+}
+
+// TestProcessSqlData is a basic test of ProcessSqlData that checks
+// 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) {
+	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{
+				{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:     "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:     "te st",
+			ColNames: []string{"a_a", "_b", "_c_"},
+			ColDefs: map[string]schema.Column{
+				"a a": schema.Column{Name: "a a", Type: schema.Type{Name: "float4"}},
+				" b":  schema.Column{Name: " b", Type: schema.Type{Name: "int8"}},
+				" 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: "te_st", cols: []string{"a a", " b", " c "}, vals: []interface{}{float64(42.3), int64(3), "cat"}},
+			spannerData{table: "te_st", cols: []string{"a a", " b", " c "}, vals: []interface{}{float64(6.6), int64(22), "dog"}},
+		},
+		rows)
+	assert.Equal(t, conv.BadRows(), int64(1))
+	assert.Equal(t, conv.SampleBadRows(10), []string{"table=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 TestConvertSqlRow_SingleCol(t *testing.T) {
+	tDate, _ := time.Parse("2006-01-02", "2019-10-29")
+	tc := []struct {
+		name    string
+		srcType schema.Type
+		spType  ddl.ScalarType
+		isArray bool
+		in      interface{} // Input value for conversion.
+		e       interface{} // Expected result.
+	}{
+		{name: "bool", srcType: schema.Type{Name: "bool"}, spType: ddl.Bool{}, in: true, e: true},
+		{name: "bool string", srcType: schema.Type{Name: "bool"}, spType: ddl.Bool{}, in: "true", e: true},
+		{name: "bytes", srcType: schema.Type{Name: "bytea"}, spType: ddl.Bytes{Len: ddl.MaxLength{}}, in: []byte{0x0, 0x1, 0xbe, 0xef}, e: []byte{0x0, 0x1, 0xbe, 0xef}},
+		{name: "date", srcType: schema.Type{Name: "date"}, spType: ddl.Date{}, in: tDate, e: getDate("2019-10-29")},
+		{name: "date string", srcType: schema.Type{Name: "date"}, spType: ddl.Date{}, in: "2019-10-29", e: getDate("2019-10-29")},
+		{name: "int64", srcType: schema.Type{Name: "bigint"}, spType: ddl.Int64{}, in: int64(42), e: int64(42)},
+		{name: "int64 string", srcType: schema.Type{Name: "text"}, spType: ddl.Int64{}, in: "42", e: int64(42)},
+		{name: "int64 float64", srcType: schema.Type{Name: "float8"}, spType: ddl.Int64{}, in: float64(42), e: int64(42)},
+		{name: "int64 byte", srcType: schema.Type{Name: "bytea"}, spType: ddl.Int64{}, in: []byte("42"), e: int64(42)},
+		{name: "float64", srcType: schema.Type{Name: "float8"}, spType: ddl.Float64{}, in: float64(42.6), e: float64(42.6)},
+		{name: "float64 string", srcType: schema.Type{Name: "text"}, spType: ddl.Float64{}, in: "42.6", e: float64(42.6)},
+		{name: "float64 int", srcType: schema.Type{Name: "bigint"}, spType: ddl.Float64{}, in: int64(42), e: float64(42)},
+		{name: "float64 byte", srcType: schema.Type{Name: "numeric"}, spType: ddl.Float64{}, in: []byte("42.6"), e: float64(42.6)},
+		{name: "string", srcType: schema.Type{Name: "text"}, spType: ddl.String{Len: ddl.MaxLength{}}, in: "eh", e: "eh"},
+		{name: "string bool", srcType: schema.Type{Name: "bool"}, spType: ddl.String{Len: ddl.MaxLength{}}, in: true, e: "true"},
+		{name: "string byte", srcType: schema.Type{Name: "bytea"}, spType: ddl.String{Len: ddl.MaxLength{}}, in: []byte("abc"), e: "abc"},
+		{name: "string int64", srcType: schema.Type{Name: "bigint"}, spType: ddl.String{Len: ddl.MaxLength{}}, in: int64(42), e: "42"},
+		{name: "string float64", srcType: schema.Type{Name: "float8"}, spType: ddl.String{Len: ddl.MaxLength{}}, in: float64(42.3), e: "42.3"},
+		{name: "string time", srcType: schema.Type{Name: "timestamp"}, spType: ddl.String{Len: ddl.MaxLength{}},
+			in: getTime(t, "2019-10-29T05:30:00+10:00"), e: "2019-10-29 05:30:00 +1000 +1000"},
+		{name: "timestamptz", srcType: schema.Type{Name: "timestamptz"}, spType: ddl.Timestamp{},
+			in: getTime(t, "2019-10-29T05:30:00+10:00"), e: getTime(t, "2019-10-29T05:30:00+10:00")},
+		{name: "timestamptz string", srcType: schema.Type{Name: "timestamptz"}, spType: ddl.Timestamp{},
+			in: "2019-10-29 05:30:00+10:00", e: getTime(t, "2019-10-29T05:30:00+10:00")},
+		{name: "timestamp", srcType: schema.Type{Name: "timestamptz"}, spType: ddl.Timestamp{},
+			in: getTime(t, "2019-10-29T05:30:00Z"), e: getTime(t, "2019-10-29T05:30:00Z")},
+		{name: "timestamp string", srcType: schema.Type{Name: "timestamptz"}, spType: ddl.Timestamp{},
+			in: "2019-10-29 05:30:00", e: getTime(t, "2019-10-29T05:30:00Z")},
+
+		// ConvertSqlRow uses convArray for conversion of array types.
+		// Since convArray is extensively tested in data_test.go, we
+		// only test a few cases here.
+		{name: "array bool", srcType: schema.Type{Name: "bool", ArrayBounds: []int64{-1}}, spType: ddl.Bool{}, isArray: true,
+			in: []byte("{true,false,NULL}"), e: []spanner.NullBool{
+				spanner.NullBool{Bool: true, Valid: true},
+				spanner.NullBool{Bool: false, Valid: true},
+				spanner.NullBool{Valid: false}}},
+		{name: "timestamp array", srcType: schema.Type{Name: "timestamptz", ArrayBounds: []int64{-1}}, spType: ddl.Timestamp{}, isArray: true,
+			in: []byte(`{"2019-10-29 05:30:00+10",NULL}`),
+			e: []spanner.NullTime{
+				spanner.NullTime{Time: getTime(t, "2019-10-29T05:30:00+10:00"), Valid: true},
+				spanner.NullTime{Valid: false}}},
+	}
+	tableName := "testtable"
+	for _, tc := range tc {
+		col := "a"
+		conv := MakeConv()
+		conv.SetLocation(time.UTC)
+		cols := []string{col}
+		srcSchema := schema.Table{Name: tableName, ColNames: []string{col}, ColDefs: map[string]schema.Column{col: schema.Column{Type: tc.srcType}}}
+		spSchema := ddl.CreateTable{
+			Name:     tableName,
+			ColNames: []string{col},
+			ColDefs:  map[string]ddl.ColumnDef{col: ddl.ColumnDef{Name: col, T: tc.spType, IsArray: tc.isArray}}}
+		ac, av, err := ConvertSqlRow(conv, tableName, cols, srcSchema, tableName, cols, spSchema, []interface{}{tc.in})
+		assert.Equal(t, cols, ac)
+		assert.Equal(t, []interface{}{tc.e}, av)
+		assert.Nil(t, err)
+	}
+}
+
+func TestConvertSqlRow_MultiCol(t *testing.T) {
+	// Tests multi-column behavior of ConvertSqlRow (including
+	// handling of null columns and synthetic keys). Also tests
+	// the combination of ProcessInfoSchema and ConvertSqlRow
+	// i.e. ConvertSqlRow 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{{"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"},
+			rows:  [][]driver.Value{}, // No primary key --> force generation of synthetic key.
+		},
+		// 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 "public"."test"`, // query is a regexp!
+			cols:  []string{"a", "b", "c"},
+			rows: [][]driver.Value{
+				{"cat", 42.3, nil},
+				{"dog", nil, 22}},
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := MakeConv()
+	err := ProcessInfoSchema(conv, db)
+	assert.Nil(t, err)
+	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", 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)}}},
+		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{{"public", "test1"}, {"public", "test2"}},
+		}, {
+			query: `SELECT COUNT[(][*][)] FROM "public"."test1"`,
+			cols:  []string{"count"},
+			rows:  [][]driver.Value{{5}},
+		}, {
+			query: `SELECT COUNT[(][*][)] FROM "public"."test2"`,
+			cols:  []string{"count"},
+			rows:  [][]driver.Value{{142}},
+		},
+	}
+	db := mkMockDB(t, ms)
+	conv := MakeConv()
+	conv.SetDataMode()
+	SetRowStats(conv, 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())
+}
+
+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
 }