Repository: cloudspannerecosystem/harbourbridge
Pull request #13 (2020-01-24 09:36:12)

Add data.
Data implements data conversion i.e. it maps PostgreSQL data to
Spanner data, based on the Spanner schema generated by schema
conversion.



Patch:
diff --git a/internal/postgres/data.go b/internal/postgres/data.go
new file mode 100644
index 0000000..bd34207
--- /dev/null
+++ b/internal/postgres/data.go
@@ -0,0 +1,267 @@
+// 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 postgres
+
+import (
+	"encoding/hex"
+	"fmt"
+	"math/bits"
+	"reflect"
+	"strconv"
+	"strings"
+	"time"
+
+	"cloud.google.com/go/civil"
+
+	"harbourbridge/spanner/ddl"
+)
+
+// ConvertData maps the PostgreSQL data in vals into Spanner data,
+// based on the specified Spanner and PostgreSQL schemas. Note that since entries
+// in vals may be empty, we also return the list of columns (empty cols are dropped).
+func ConvertData(conv *Conv, spTable, pgTable string, cols []string, vals []string) ([]string, []interface{}, error) {
+	spannerSchema, ok1 := conv.spSchema[spTable]
+	pgSchema, ok2 := conv.pgSchema[pgTable]
+	if !ok1 || !ok2 {
+		return []string{}, []interface{}{}, fmt.Errorf("can't find table %s in schema", spTable)
+	}
+	var c []string
+	var v []interface{}
+	if len(cols) != len(vals) {
+		return []string{}, []interface{}{}, fmt.Errorf("bad parameters: cols and vals have different lengths: len(cols)=%d, len(vals)=%d", len(cols), len(vals))
+	}
+	for i, col := range cols {
+		if vals[i] == "\\N" { // PostgreSQL representation of empty column in COPY-FROM blocks.
+			continue
+		}
+		cd, ok := spannerSchema.Cds[col]
+		if !ok {
+			return []string{}, []interface{}{}, fmt.Errorf("can't find col %s in schema", col)
+		}
+		pgCol, ok := pgSchema.cols[col]
+		if !ok {
+			return []string{}, []interface{}{}, fmt.Errorf("can't find col %s in pgSchema", col)
+		}
+		var x interface{}
+		var err error
+		if cd.IsArray {
+			x, err = convArray(cd.T, pgCol, conv.location, vals[i])
+		} else {
+			x, err = convScalar(cd.T, pgCol.id, conv.location, vals[i])
+		}
+		if err != nil {
+			return []string{}, []interface{}{}, err
+		}
+		v = append(v, x)
+		c = append(c, col)
+	}
+	if aux, ok := conv.syntheticPKeys[spTable]; ok {
+		c = append(c, aux.col)
+		v = append(v, int64(bits.Reverse64(uint64(aux.sequence))))
+		aux.sequence++
+		conv.syntheticPKeys[spTable] = aux
+	}
+	return c, v, nil
+}
+
+func convScalar(spannerType ddl.ScalarType, pgType string, location *time.Location, val string) (interface{}, error) {
+	// Whitespace within the val string is considered part of the data value.
+	// Note that many of the underlying conversions functions we use (like
+	// strconv.ParseFloat and strconv.ParseInt) return "invalid syntax"
+	// errors if whitespace were to appear at the start or end of a string.
+	// We do not expect pg_dump to generate such output.
+	switch spannerType.(type) {
+	case ddl.Bool:
+		return convBool(val)
+	case ddl.Bytes:
+		return convBytes(val)
+	case ddl.Date:
+		return convDate(val)
+	case ddl.Float64:
+		return convFloat64(val)
+	case ddl.Int64:
+		return convInt64(val)
+	case ddl.String:
+		return val, nil
+	case ddl.Timestamp:
+		return convTimestamp(pgType, location, val)
+	default:
+		return val, fmt.Errorf("data conversion not implemented for type %v", reflect.TypeOf(spannerType))
+	}
+}
+
+func convBool(val string) (bool, error) {
+	b, err := strconv.ParseBool(val)
+	if err != nil {
+		return b, fmt.Errorf("can't convert to bool: %w", err)
+	}
+	return b, err
+}
+
+func convBytes(val string) ([]byte, error) {
+	if val[0:3] != `\\x` {
+		return []byte{}, fmt.Errorf("can't convert to bytes: doesn't start with \\x prefix")
+	}
+	b, err := hex.DecodeString(val[3:])
+	if err != nil {
+		return b, fmt.Errorf("can't convert to bytes: %w", err)
+	}
+	return b, err
+}
+
+func convDate(val string) (civil.Date, error) {
+	d, err := civil.ParseDate(val)
+	if err != nil {
+		return d, fmt.Errorf("can't convert to date: %w", err)
+	}
+	return d, err
+}
+
+func convFloat64(val string) (float64, error) {
+	f, err := strconv.ParseFloat(val, 64)
+	if err != nil {
+		return f, fmt.Errorf("can't convert to float64: %w", err)
+	}
+	return f, err
+}
+
+func convInt64(val string) (int64, error) {
+	i, err := strconv.ParseInt(val, 10, 64)
+	if err != nil {
+		return i, fmt.Errorf("can't convert to int64: %w", err)
+	}
+	return i, err
+}
+
+// convTimestamp maps a PostgreSQL timestamp into a go Time (which
+// is translated to a Spanner timestamp by the go Spanner client library).
+// It handles both timestamptz and timestamp conversions.
+// Note that PostgreSQL supports a wide variety of different timestamp
+// formats (see https://www.postgresql.org/docs/9.1/datatype-datetime.html).
+// We don't attempt to support all of these timestamp formats. Our goal
+// is more modest: we just need to support the formats generated by
+// pg_dump.
+func convTimestamp(pgType string, location *time.Location, val string) (t time.Time, err error) {
+	// 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 pgType == "timestamptz" {
+		// PostgreSQL abbreviates timezone to just hour where possible.
+		t, err = time.Parse("2006-01-02 15:04:05Z07", val)
+		if err != nil {
+			// Try using hour and min for timezone e.g. PGTZ set to 'Asia/Kolkata'.
+			t, err = time.Parse("2006-01-02 15:04:05Z07:00", val)
+		}
+		if err != nil {
+			// Try parsing without timezone. Some pg_dump files
+			// generate timestamps without timezone for timestampz data
+			// e.g. the Pagila port of Sakila. We interpret these timestamps
+			// using the current time location (default is local time).
+			// Note: we might want to look for "SET TIME ZONE" in the pg_dump
+			// and interpret wrt that timezone.
+			t, err = time.ParseInLocation("2006-01-02 15:04:05", val, location)
+		}
+	} else {
+		// timestamp without time zone: data should just consist of date and time.
+		// timestamp conversion should ignore timezone. We mimic this using Parse
+		// i.e. treat it as UTC, so it will be stored 'as-is' in Spanner.
+		t, err = time.Parse("2006-01-02 15:04:05", val)
+	}
+	if err != nil {
+		return t, fmt.Errorf("can't convert to timestamp (posgres type: %s)", pgType)
+	}
+	return t, err
+}
+
+func convArray(spannerType ddl.ScalarType, pgType pgColDef, location *time.Location, v string) (interface{}, error) {
+	v = strings.TrimSpace(v)
+	if v[0] != '{' || v[len(v)-1] != '}' {
+		return []interface{}{}, fmt.Errorf("unrecognized data format for array: expected {v1, v2, ...}")
+	}
+	a := strings.Split(v[1:len(v)-1], ",")
+
+	// The Spanner client for go does not accept []interface{} for arrays.
+	// Instead it only accepts slices of a specific type e.g. []int64, []string.
+	// Hence we have to do the following case analysis.
+	switch spannerType.(type) {
+	case ddl.Bool:
+		var r []bool
+		for _, s := range a {
+			b, err := convBool(s)
+			if err != nil {
+				return []bool{}, err
+			}
+			r = append(r, b)
+		}
+		return r, nil
+	case ddl.Bytes:
+		var r [][]byte
+		for _, s := range a {
+			b, err := convBytes(s)
+			if err != nil {
+				return [][]byte{}, err
+			}
+			r = append(r, b)
+		}
+		return r, nil
+	case ddl.Date:
+		var r []civil.Date
+		for _, s := range a {
+			d, err := convDate(s)
+			if err != nil {
+				return []civil.Date{}, err
+			}
+			r = append(r, d)
+		}
+		return r, nil
+	case ddl.Float64:
+		var r []float64
+		for _, s := range a {
+			f, err := convFloat64(s)
+			if err != nil {
+				return []float64{}, err
+			}
+			r = append(r, f)
+		}
+		return r, nil
+	case ddl.Int64:
+		var r []int64
+		for _, s := range a {
+			i, err := convInt64(s)
+			if err != nil {
+				return r, err
+			}
+			r = append(r, i)
+		}
+		return r, nil
+	case ddl.String:
+		var r []string
+		for _, s := range a {
+			r = append(r, s)
+		}
+		return r, nil
+	case ddl.Timestamp:
+		var r []time.Time
+		for _, s := range a {
+			t, err := convTimestamp(pgType.id, location, s)
+			if err != nil {
+				return []time.Time{}, err
+			}
+			r = append(r, t)
+		}
+		return r, nil
+	}
+	return []interface{}{}, fmt.Errorf("array type conversion not implemented for type %v", reflect.TypeOf(spannerType))
+}
diff --git a/internal/postgres/tablemap.go b/internal/postgres/tablemap.go
new file mode 100644
index 0000000..5bb3ad8
--- /dev/null
+++ b/internal/postgres/tablemap.go
@@ -0,0 +1,117 @@
+// 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 postgres
+
+import (
+	"fmt"
+	"strconv"
+
+	"harbourbridge/internal"
+)
+
+// GetSpannerTable maps a PostgreSQL table name into a legal Spanner table
+// name. Note that PostgreSQL column names can be essentially any string, but
+// Spanner column names must use a limited character set. This means that
+// getSpannerTable may have to change a name to make it legal, we must ensure
+// that:
+// a) the new table name is legal
+// b) the new table name doesn't clash with other Spanner table names
+// c) we consistently return the same name for this table.
+func GetSpannerTable(conv *Conv, pgTable string) (string, error) {
+	if pgTable == "" {
+		return "", fmt.Errorf("bad parameter: table string is empty")
+	}
+	if sp, found := conv.toSpanner[pgTable]; found {
+		return sp.name, nil
+	}
+	spTable, _ := internal.FixName(pgTable)
+	if _, found := conv.toPostgres[spTable]; found {
+		// s has been used before i.e. FixName caused a collision.
+		// Add unique postfix: use number of tables so far.
+		// However, there is a chance this has already been used,
+		// so need to iterate.
+		id := len(conv.toSpanner)
+		for {
+			t := spTable + "_" + strconv.Itoa(id)
+			if _, found := conv.toPostgres[t]; !found {
+				spTable = t
+				break
+			}
+			id++
+		}
+	}
+	if spTable != pgTable {
+		internal.VerbosePrintf("Mapping PostgreSQL table %s to Spanner table %s\n", pgTable, spTable)
+	}
+	conv.toSpanner[pgTable] = nameAndCols{name: spTable, cols: make(map[string]string)}
+	conv.toPostgres[spTable] = nameAndCols{name: pgTable, cols: make(map[string]string)}
+	return spTable, nil
+}
+
+// GetSpannerCol maps a PostgreSQL table/column into a legal Spanner column
+// name. If mustExist is true, we return error if the column is new.
+// Note that PostgreSQL column names can be essentially any string, but
+// Spanner column names must use a limited character set. This means that
+// getSpannerCol may have to change a name to make it legal, we must ensure
+// that:
+// a) the new col name is legal
+// b) the new col name doesn't clash with other col names in the same table
+// c) we consistently return the same name for the same col.
+func GetSpannerCol(conv *Conv, pgTable, pgCol string, mustExist bool) (string, error) {
+	if pgTable == "" {
+		return "", fmt.Errorf("bad parameter: table string is empty")
+	}
+	if pgCol == "" {
+		return "", fmt.Errorf("bad parameter: col string is empty")
+	}
+	sp, found := conv.toSpanner[pgTable]
+	if !found {
+		return "", fmt.Errorf("unknown table %s", pgTable)
+	}
+	// Sanity check: do reverse mapping and check consistency.
+	// Consider dropping this check.
+	pg, found := conv.toPostgres[sp.name]
+	if !found || pg.name != pgTable {
+		return "", fmt.Errorf("internal error: table mapping inconsistency for table %s (%s)", pgTable, pg.name)
+	}
+	if spCol, found := sp.cols[pgCol]; found {
+		return spCol, nil
+	}
+	if mustExist {
+		return "", fmt.Errorf("table %s does not have a column %s", pgTable, pgCol)
+	}
+	spCol, _ := internal.FixName(pgCol)
+	if _, found := conv.toPostgres[sp.name].cols[spCol]; found {
+		// spCol has been used before i.e. FixName caused a collision.
+		// Add unique postfix: use number of cols in this table so far.
+		// However, there is a chance this has already been used,
+		// so need to iterate.
+		id := len(sp.cols)
+		for {
+			c := spCol + "_" + strconv.Itoa(id)
+			if _, found := conv.toPostgres[sp.name].cols[c]; !found {
+				spCol = c
+				break
+			}
+			id++
+		}
+	}
+	if spCol != pgCol {
+		internal.VerbosePrintf("Mapping PostgreSQL col %s (table %s) to Spanner col %s\n", pgCol, pgTable, spCol)
+	}
+	conv.toSpanner[pgTable].cols[pgCol] = spCol
+	conv.toPostgres[sp.name].cols[spCol] = pgCol
+	return spCol, nil
+}


Test patch:
diff --git a/internal/postgres/data_test.go b/internal/postgres/data_test.go
new file mode 100644
index 0000000..80d0efe
--- /dev/null
+++ b/internal/postgres/data_test.go
@@ -0,0 +1,238 @@
+// 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 postgres
+
+import (
+	"fmt"
+	"math/bits"
+	"testing"
+	"time"
+
+	"cloud.google.com/go/civil"
+	"github.com/stretchr/testify/assert"
+
+	"harbourbridge/spanner/ddl"
+)
+
+func TestConvertData(t *testing.T) {
+	singleColTests := []struct {
+		name    string
+		t       ddl.ScalarType
+		isArray bool
+		pgt     string      // Postgres type (used by e.g. timestamp conversions).
+		in      string      // Input value for conversion.
+		e       interface{} // Expected result.
+	}{
+		{"bool", ddl.Bool{}, false, "", "true", true},
+		{"bytes", ddl.Bytes{ddl.MaxLength{}}, false, "", `\\x0001beef`, []byte{0x0, 0x1, 0xbe, 0xef}},
+		{"date", ddl.Date{}, false, "", "2019-10-29", getDate("2019-10-29")},
+		{"float64", ddl.Float64{}, false, "", "42.6", float64(42.6)},
+		{"int64", ddl.Int64{}, false, "", "42", int64(42)},
+		{"string", ddl.String{ddl.MaxLength{}}, false, "", "eh", "eh"},
+		{"timestamptz", ddl.Timestamp{}, false, "timestamptz", "2019-10-29 05:30:00+10", getTime(t, "2019-10-29T05:30:00+10:00")},
+		{"timestamp", ddl.Timestamp{}, false, "timestamp", "2019-10-29 05:30:00", getTime(t, "2019-10-29T05:30:00Z")},
+		// Add cases for each array type, since each is a separate code path.
+		{"bool array", ddl.Bool{}, true, "", "{true,false}", []bool{true, false}},
+		{"bytes array", ddl.Bytes{ddl.MaxLength{}}, true, "", `{\\x0001beef}`, [][]byte{{0x0, 0x1, 0xbe, 0xef}}},
+		{"date array", ddl.Date{}, true, "", "{2019-10-29,2019-10-28}", []civil.Date{getDate("2019-10-29"), getDate("2019-10-28")}},
+		{"float64 array", ddl.Float64{}, true, "", "{1.1,2.2,3.3}", []float64{1.1, 2.2, 3.3}},
+		{"int64 array", ddl.Int64{}, true, "", "{1,2,3}", []int64{1, 2, 3}},
+		{"string array", ddl.String{ddl.MaxLength{}}, true, "", "{1,2,3}", []string{"1", "2", "3"}},
+		{"timestamp array", ddl.Timestamp{}, true, "timestamptz", "{2019-10-29 05:30:00+10}", []time.Time{getTime(t, "2019-10-29T05:30:00+10:00")}},
+	}
+	for _, tc := range singleColTests {
+		table := "testtable"
+		col := "a"
+		conv := buildConv(table,
+			ddl.CreateTable{table, []string{col}, map[string]ddl.ColumnDef{col: ddl.ColumnDef{col, tc.t, tc.isArray, false, ""}}, []ddl.IndexKey{}, ""},
+			pgTableDef{map[string]pgColDef{col: pgColDef{id: tc.pgt}}})
+		conv.SetLocation(time.UTC)
+		ac, av, err := ConvertData(conv, table, table, []string{col}, []string{tc.in})
+		checkResults(t, ac, av, err, []string{col}, []interface{}{tc.e}, tc.name)
+	}
+
+	timestampTests := []struct {
+		name string
+		pgt  string
+		in   string
+		e    interface{}
+	}{
+		{"timestamptz hour timezone", "timestamptz", "2019-10-29 05:30:00+10:00", getTime(t, "2019-10-29T05:30:00+10:00")},
+		{"timestamptz hour/min timezone", "timestamptz", "2019-10-29 05:30:00+10:30", getTime(t, "2019-10-29T05:30:00+10:30")},
+		{"timestamptz no timezone", "timestamptz", "2019-10-29 05:30:00", getTime(t, "2019-10-29T05:30:00+11:00")},
+		{"timestamp", "timestamp", "2019-10-29 05:30:00", getTime(t, "2019-10-29T05:30:00Z")},
+	}
+	for _, tc := range timestampTests {
+		table := "testtable"
+		col := "a"
+		conv := buildConv(table,
+			ddl.CreateTable{table, []string{col}, map[string]ddl.ColumnDef{col: ddl.ColumnDef{Name: col, T: ddl.Timestamp{}}}, []ddl.IndexKey{}, ""},
+			pgTableDef{map[string]pgColDef{col: pgColDef{id: tc.pgt}}})
+		loc, _ := time.LoadLocation("Australia/Sydney")
+		conv.SetLocation(loc) // Set location so test is robust i.e. doesn't depent on local timezone.
+		ac, av, err := ConvertData(conv, table, table, []string{col}, []string{tc.in})
+		assert.Nil(t, err, tc.name)
+		assert.Equal(t, []string{col}, ac, tc.name+": column mismatch")
+		// Avoid assert.Equal for time.Time (it forces location equality).
+		// Instead use Time.Equals, which determines equality based on whether
+		// two times represent the same instant.
+		assert.Equal(t, 1, len(av))
+		at, ok1 := av[0].(time.Time)
+		et, ok2 := tc.e.(time.Time)
+		assert.True(t, ok1 && ok2, tc.name+": cast to Time failed")
+		assert.True(t, at.Equal(et), tc.name+": value mismatch")
+	}
+
+	multiColTests := []struct {
+		name  string
+		cols  []string      // Input columns.
+		vals  []string      // Input values.
+		ecols []string      // Expected columns.
+		evals []interface{} // Expected values.
+	}{
+		{
+			name:  "Cols in order",
+			cols:  []string{"a", "b", "c"},
+			vals:  []string{"6", "6.6", "true"},
+			ecols: []string{"a", "b", "c"},
+			evals: []interface{}{int64(6), float64(6.6), true},
+		},
+		{
+			name:  "Cols out of order",
+			cols:  []string{"b", "c", "a"},
+			vals:  []string{"6.6", "true", "6"},
+			ecols: []string{"b", "c", "a"},
+			evals: []interface{}{float64(6.6), true, int64(6)},
+		},
+		{
+			name:  "Null column",
+			cols:  []string{"a", "b", "c"},
+			vals:  []string{"6", "\\N", "true"},
+			ecols: []string{"a", "c"},
+			evals: []interface{}{int64(6), true},
+		},
+		{
+			name:  "Missing columns",
+			cols:  []string{"a"},
+			vals:  []string{"6"},
+			ecols: []string{"a"},
+			evals: []interface{}{int64(6)},
+		},
+	}
+	spSchema := ddl.CreateTable{
+		"testtable",
+		[]string{"a", "b", "c"},
+		map[string]ddl.ColumnDef{
+			"a": ddl.ColumnDef{Name: "a", T: ddl.Int64{}},
+			"b": ddl.ColumnDef{Name: "b", T: ddl.Float64{}},
+			"c": ddl.ColumnDef{Name: "c", T: ddl.Bool{}},
+		},
+		[]ddl.IndexKey{},
+		"",
+	}
+	pgSchema := pgTableDef{
+		map[string]pgColDef{
+			"a": pgColDef{id: "int8"},
+			"b": pgColDef{id: "float8"},
+			"c": pgColDef{id: "bool"},
+		}}
+	for _, tc := range multiColTests {
+		conv := buildConv(spSchema.Name, spSchema, pgSchema)
+		acols, avals, err := ConvertData(conv, spSchema.Name, spSchema.Name, tc.cols, tc.vals)
+		checkResults(t, acols, avals, err, tc.ecols, tc.evals, tc.name)
+	}
+
+	errorTests := []struct {
+		name string
+		cols []string // Input columns.
+		vals []string // Input values.
+	}{
+		{
+			name: "Error in int64",
+			cols: []string{"a", "b", "c"},
+			vals: []string{" 6", "6.6", "true"},
+		},
+		{
+			name: "Error in float64",
+			cols: []string{"a", "b", "c"},
+			vals: []string{"6", "6.6e", "true"},
+		},
+		{
+			name: "Error in bool",
+			cols: []string{"a", "b", "c"},
+			vals: []string{"6", "6.6", "truee"},
+		},
+	}
+	for _, tc := range errorTests {
+		conv := buildConv(spSchema.Name, spSchema, pgSchema)
+		acols, avals, err := ConvertData(conv, spSchema.Name, spSchema.Name, tc.cols, tc.vals)
+		assert.NotNil(t, err, tc.name)
+		assert.Equal(t, []string{}, acols, tc.name+": column mismatch")
+		assert.Equal(t, []interface{}{}, avals, tc.name+": value mismatch")
+	}
+
+	syntheticPKeyTests := []struct {
+		name  string
+		cols  []string      // Input columns.
+		vals  []string      // Input values.
+		ecols []string      // Expected columns.
+		evals []interface{} // Expected values.
+	}{
+		{
+			name:  "Sequence 0",
+			cols:  []string{"a", "b", "c"},
+			vals:  []string{"6", "6.6", "true"},
+			ecols: []string{"a", "b", "c", "synth_id"},
+			evals: []interface{}{int64(6), float64(6.6), true, int64(0)},
+		},
+		{
+			name:  "Sequence 1",
+			cols:  []string{"a"},
+			vals:  []string{"7"},
+			ecols: []string{"a", "synth_id"},
+			evals: []interface{}{int64(7), int64(bits.Reverse64(1))},
+		},
+	}
+	conv := buildConv(spSchema.Name, spSchema, pgSchema)
+	conv.syntheticPKeys[spSchema.Name] = syntheticPKey{col: "synth_id", sequence: 0}
+	for _, tc := range syntheticPKeyTests {
+		acols, avals, err := ConvertData(conv, spSchema.Name, spSchema.Name, tc.cols, tc.vals)
+		checkResults(t, acols, avals, err, tc.ecols, tc.evals, tc.name)
+	}
+}
+
+func buildConv(table string, spSchema ddl.CreateTable, pgSchema pgTableDef) *Conv {
+	conv := MakeConv()
+	conv.spSchema[table] = spSchema
+	conv.pgSchema[table] = pgSchema
+	return conv
+}
+
+func checkResults(t *testing.T, acols []string, avals []interface{}, err error, ecols []string, evals []interface{}, name string) {
+	assert.Nil(t, err, name)
+	assert.Equal(t, ecols, acols, name+": column mismatch")
+	assert.Equal(t, evals, avals, name+": value mismatch")
+}
+
+func getTime(t *testing.T, s string) time.Time {
+	x, err := time.Parse(time.RFC3339, s)
+	assert.Nil(t, err, fmt.Sprintf("getTime can't parse %s:", s))
+	return x
+}
+
+func getDate(s string) civil.Date {
+	d, _ := civil.ParseDate(s)
+	return d
+}
diff --git a/internal/postgres/tablemap_test.go b/internal/postgres/tablemap_test.go
new file mode 100644
index 0000000..e057bb5
--- /dev/null
+++ b/internal/postgres/tablemap_test.go
@@ -0,0 +1,93 @@
+// 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 postgres
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetSpannerTable(t *testing.T) {
+	conv := MakeConv()
+	basicTests := []struct {
+		name    string // Name of test.
+		pgTable string // PostgreSQL table name to test.
+		error   bool   // Whether an error is expected.
+		spTable string // Expected Spanner table name.
+	}{
+		{"Empty", "", true, ""},
+		{"Good name", "table", false, "table"},
+		{"Good name: setup collision", "tab_le_5", false, "tab_le_5"},
+		{"Good name: setup collision (2)", "tab_le_6", false, "tab_le_6"},
+		{"Illegal character", "tab\nle", false, "tab_le"},
+		{"Illegal character with collision (1)", "tab\tle", false, "tab_le_4"},
+		{"Illegal character with collision (2)", "tab?le", false, "tab_le_7"}, // Must skip tab_le_5 and tab_le_6.
+		{"Collision with previous remapping", "tab_le_4", false, "tab_le_4_6"},
+		{"Illegal start character", "2table", false, "Atable"},
+		{"Illegal start character with collision (1)", "_table", false, "Atable_8"},
+		{"Illegal start character with collision (2)", "\ntable", false, "Atable_9"},
+	}
+	for _, tc := range basicTests {
+		spTable, err := GetSpannerTable(conv, tc.pgTable)
+		if tc.error {
+			assert.NotNil(t, err, tc.name)
+			continue
+		}
+		assert.Equal(t, tc.spTable, spTable, tc.name)
+		// Run again to check we get same result.
+		s2, err := GetSpannerTable(conv, tc.pgTable)
+		assert.Nil(t, err, tc.name)
+		assert.Equal(t, spTable, s2, tc.name)
+	}
+}
+
+func TestGetSpannerCol(t *testing.T) {
+	conv := MakeConv()
+	basicTests := []struct {
+		name    string // Name of test.
+		pgTable string // PostgreSQL table name to test.
+		pgCol   string // PostgreSQL col name to test.
+		error   bool   // Whether an error is expected.
+		spCol   string // Expected Spanner column name.
+	}{
+		{"Empty table", "", "col", true, ""},
+		{"Empty col", "table", "", true, ""},
+		{"Good name", "table", "col", false, "col"},
+		{"Bad table", "ta.b\nle", "col", false, "col"},
+		{"Bad col", "table", "c\nol", false, "c_ol"},
+		{"Bad table and col", "t.able", "c\no\nl", false, "c_o_l"},
+		{"table1 good name 1", "table1", "col", false, "col"},
+		{"table1 good name 2", "table1", "c_ol", false, "c_ol"},
+		{"table1 good name 3", "table1", "c_ol_5", false, "c_ol_5"},
+		{"table1 good name 4", "table1", "c_ol_6", false, "c_ol_6"},
+		{"table1 collision 1", "table1", "c\tol", false, "c_ol_4"},
+		{"table1 collision 2", "table1", "c\nol", false, "c_ol_7"}, // Skip c_ol_5 and c_ol_6.
+		{"table1 collision 3", "table1", "c?ol", false, "c_ol_8"},
+	}
+	for _, tc := range basicTests {
+		_, err1 := GetSpannerTable(conv, tc.pgTable) // Ensure table is known.
+		spCol, err2 := GetSpannerCol(conv, tc.pgTable, tc.pgCol, false)
+		if tc.error {
+			assert.True(t, err1 != nil || err2 != nil, tc.name)
+			continue
+		}
+		assert.Equal(t, tc.spCol, spCol, tc.name)
+		// Run again to check we get same result.
+		spCol2, err := GetSpannerCol(conv, tc.pgTable, tc.pgCol, false)
+		assert.Nil(t, err, tc.name)
+		assert.Equal(t, spCol, spCol2, tc.name)
+	}
+}