Repository: cloudspannerecosystem/harbourbridge
Pull request #48 (2020-02-23 23:48:03)

Add schema and schema-to-ddl conversion.
Adds a generic schema representation, and conversion from this schema
to Spanner DDL. Also adds a test and enough changes to convert.go to
get this to compile (add new fields to Cvt).


Patch:
diff --git a/internal/postgres/convert.go b/internal/convert.go
similarity index 73%
rename from internal/postgres/convert.go
rename to internal/convert.go
index fa5c732..7df79a1 100644
--- a/internal/postgres/convert.go
+++ b/internal/convert.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"fmt"
@@ -22,18 +22,19 @@ import (
 
 	nodes "github.com/lfittl/pg_query_go/nodes"
 
-	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
 // Conv contains all schema and data conversion state.
 type Conv struct {
-	mode           mode                       // Schema mode or data mode.
-	spSchema       map[string]ddl.CreateTable // Maps Spanner table name to Spanner schema.
-	syntheticPKeys map[string]syntheticPKey   // Maps Spanner table name to synthetic primary key (if needed).
-	pgSchema       map[string]pgTableDef      // Maps PostgreSQL table name Postgres schema information.
-	toSpanner      map[string]nameAndCols     // Maps from PostgreSQL table name to Spanner name and column mapping.
-	toPostgres     map[string]nameAndCols     // Maps from Spanner table name to PostgreSQL name and column mapping.
+	mode           mode                                // Schema mode or data mode.
+	spSchema       map[string]ddl.CreateTable          // Maps Spanner table name to Spanner schema.
+	syntheticPKeys map[string]syntheticPKey            // Maps Spanner table name to synthetic primary key (if needed).
+	srcSchema      map[string]schema.Table             // Maps source-DB table name to schema information.
+	issues         map[string]map[string][]schemaIssue // Maps source-DB table/col to list of schema conversion issues.
+	toSpanner      map[string]nameAndCols              // Maps from source-DB table name to Spanner name and column mapping.
+	toSource       map[string]nameAndCols              // Maps from Spanner table name to source-DB table name and column mapping.
 	dataSink       func(table string, cols []string, values []interface{})
 	location       *time.Location // Timezone (for timestamp conversion).
 	sampleBadRows  rowSamples     // Rows that generated errors during conversion.
@@ -49,31 +50,17 @@ const (
 
 // syntheticPKey specifies a synthetic primary key and current sequence
 // count for a table, if needed. We use a synthetic primary key when
-// a PostgreSQL table has no primary key.
+// the source DB table has no primary key.
 type syntheticPKey struct {
 	col      string
 	sequence int64
 }
 
-// pgTableDef captures data about a table's PostgreSQL schema.
-// Note: we only keep a minimal set of PostgreSQL schema information.
-type pgTableDef struct {
-	cols map[string]pgColDef
-}
-
-// pgColDef collects key PostgreSQL schema parameters for a table column.
-type pgColDef struct {
-	id     string        // Type id.
-	mods   []int64       // List of modifiers (aka type parameters e.g. varchar(8) or numeric(6, 4).
-	array  []int64       // Array bound information. Empty for scalar types.
-	issues []schemaIssue // List of issues encountered mapping this col to Spanner.
-}
-
 type schemaIssue int
 
 // Defines all of the schema issues we track. Includes issues
-// with type mappings, as well as features (such as PostgreSQL
-// constraints) that aren't supported in Spanner.
+// with type mappings, as well as features (such as source
+// DB constraints) that aren't supported in Spanner.
 const (
 	defaultValue schemaIssue = iota
 	foreignKey
@@ -88,7 +75,7 @@ const (
 )
 
 // nameAndCols contains the name of a table and its columns.
-// Used to map between PostgreSQL and Spanner table and column names.
+// Used to map between source DB and Spanner table and column names.
 type nameAndCols struct {
 	name string
 	cols map[string]string
@@ -113,9 +100,9 @@ type row struct {
 // c) successfully converted, but an error occurs when writing the row to Spanner.
 // d) unsuccessfully converted (we won't try to write such rows to Spanner).
 type stats struct {
-	rows       map[string]int64          // Count of rows encountered during processing (a + b + c + d), broken down by Spanner table.
-	goodRows   map[string]int64          // Count of rows successfully converted (b + c), broken down by Spanner table.
-	badRows    map[string]int64          // Count of rows where conversion failed (d), broken down by Spanner table.
+	rows       map[string]int64          // Count of rows encountered during processing (a + b + c + d), broken down by source table.
+	goodRows   map[string]int64          // Count of rows successfully converted (b + c), broken down by source table.
+	badRows    map[string]int64          // Count of rows where conversion failed (d), broken down by source table.
 	statement  map[string]*statementStat // Count of processed statements, broken down by statement type.
 	unexpected map[string]int64          // Count of unexpected conditions, broken down by condition description.
 	reparsed   int64                     // Count of times we re-parse pg_dump data looking for end-of-statement.
@@ -133,9 +120,10 @@ func MakeConv() *Conv {
 	return &Conv{
 		spSchema:       make(map[string]ddl.CreateTable),
 		syntheticPKeys: make(map[string]syntheticPKey),
-		pgSchema:       make(map[string]pgTableDef),
+		srcSchema:      make(map[string]schema.Table),
+		issues:         make(map[string]map[string][]schemaIssue),
 		toSpanner:      make(map[string]nameAndCols),
-		toPostgres:     make(map[string]nameAndCols),
+		toSource:       make(map[string]nameAndCols),
 		location:       time.Local, // By default, use go's local time, which uses $TZ (when set).
 		sampleBadRows:  rowSamples{bytesLimit: 10 * 1000 * 1000},
 		stats: stats{
@@ -246,8 +234,8 @@ func (conv *Conv) AddPrimaryKeys() {
 	for t, ct := range conv.spSchema {
 		if len(ct.Pks) == 0 {
 			k := conv.buildPrimaryKey(t)
-			ct.Cols = append(ct.Cols, k)
-			ct.Cds[k] = ddl.ColumnDef{Name: k, T: ddl.Int64{}}
+			ct.ColNames = append(ct.ColNames, k)
+			ct.ColDefs[k] = ddl.ColumnDef{Name: k, T: ddl.Int64{}}
 			ct.Pks = []ddl.IndexKey{ddl.IndexKey{Col: k}}
 			conv.spSchema[t] = ct
 			conv.syntheticPKeys[t] = syntheticPKey{k, 0}
@@ -262,15 +250,15 @@ func (conv *Conv) SetLocation(loc *time.Location) {
 
 func (conv *Conv) buildPrimaryKey(spTable string) string {
 	base := "synth_id"
-	if _, ok := conv.toPostgres[spTable]; !ok {
-		conv.unexpected(fmt.Sprintf("toPostgres lookup fails for table %s: ", spTable))
+	if _, ok := conv.toSource[spTable]; !ok {
+		conv.unexpected(fmt.Sprintf("toSource lookup fails for table %s: ", spTable))
 		return base
 	}
 	count := 0
 	key := base
 	for {
 		// Check key isn't already a column in the table.
-		if _, ok := conv.toPostgres[spTable].cols[key]; !ok {
+		if _, ok := conv.toSource[spTable].cols[key]; !ok {
 			return key
 		}
 		key = fmt.Sprintf("%s%d", base, count)
@@ -283,7 +271,7 @@ func (conv *Conv) buildPrimaryKey(spTable string) string {
 // be completely reliable due to potential double-counting
 // because we process pg_dump data twice.
 func (conv *Conv) unexpected(u string) {
-	internal.VerbosePrintf("Unexpected condition: %s\n", u)
+	VerbosePrintf("Unexpected condition: %s\n", u)
 	// Limit size of unexpected map. If over limit, then only
 	// update existing entries.
 	if _, ok := conv.stats.unexpected[u]; ok || len(conv.stats.unexpected) < 1000 {
@@ -291,32 +279,33 @@ func (conv *Conv) unexpected(u string) {
 	}
 }
 
-// statsAddRow increments the rows stats for table 'spTable' if b is true.
-// This is used to avoid double counting of stats. Specifically, some code paths
-// that report row stats run in both schema-mode and data-mode e.g. statement.go.
-// To avoid double counting, we explicitly choose a mode-for-stats-collection
-// for each place where row stats are collected. When specifying this mode
-// Take care to ensure that the code actually runs in the mode you specify,
+// statsAddRow increments the count of rows for 'srcTable' if b is
+// true.  The boolean arg 'b' is used to avoid double counting of
+// stats. Specifically, some code paths that report row stats run in
+// both schema-mode and data-mode e.g. statement.go.  To avoid double
+// counting, we explicitly choose a mode-for-stats-collection for each
+// place where row stats are collected. When specifying this mode take
+// care to ensure that the code actually runs in the mode you specify,
 // otherwise stats will be dropped.
-func (conv *Conv) statsAddRow(spTable string, b bool) {
+func (conv *Conv) statsAddRow(srcTable string, b bool) {
 	if b {
-		conv.stats.rows[spTable]++
+		conv.stats.rows[srcTable]++
 	}
 }
 
-// statsAddGoodRow increments the good-row stats for table 'spTable' if b is true.
-// See statsAddRow comments for context.
-func (conv *Conv) statsAddGoodRow(spTable string, b bool) {
+// 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) {
 	if b {
-		conv.stats.goodRows[spTable]++
+		conv.stats.goodRows[srcTable]++
 	}
 }
 
-// statsAddBadRow increments the bad-row stats for table 'spTable' if b is true.
-// See statsAddRow comments for context.
-func (conv *Conv) statsAddBadRow(spTable string, b bool) {
+// 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) {
 	if b {
-		conv.stats.badRows[spTable]++
+		conv.stats.badRows[srcTable]++
 	}
 }
 
@@ -343,7 +332,7 @@ func (conv *Conv) getStatementStat(s string) *statementStat {
 func (conv *Conv) skipStatement(l []nodes.Node) {
 	if conv.schemaMode() { // Record statement stats on first pass only.
 		s := prNodes(l)
-		internal.VerbosePrintf("Skipping statement: %s\n", s)
+		VerbosePrintf("Skipping statement: %s\n", s)
 		conv.getStatementStat(s).skip++
 	}
 }
@@ -351,7 +340,7 @@ func (conv *Conv) skipStatement(l []nodes.Node) {
 func (conv *Conv) errorInStatement(l []nodes.Node) {
 	if conv.schemaMode() { // Record statement stats on first pass only.
 		s := prNodes(l)
-		internal.VerbosePrintf("Error processing statement: %s\n", s)
+		VerbosePrintf("Error processing statement: %s\n", s)
 		conv.getStatementStat(s).error++
 	}
 }
diff --git a/internal/postgres/data.go b/internal/data.go
similarity index 62%
rename from internal/postgres/data.go
rename to internal/data.go
index 14839fd..1732b13 100644
--- a/internal/postgres/data.go
+++ b/internal/data.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"encoding/hex"
@@ -28,44 +28,85 @@ import (
 	"github.com/cloudspannerecosystem/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]
+// ProcessDataRow converts a row of data and writes it out to Spanner.
+// srcTable and srcCols are the source table and columns respectively,
+// and vals contains string data to be converted to appropriate types
+// to send to Spanner.  ProcessDataRow is only called in dataMode.
+func ProcessDataRow(conv *Conv, srcTable string, srcCols, vals []string) {
+	spTable, spCols, spVals, err := ConvertData(conv, srcTable, srcCols, vals)
+	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
+		}
+	} 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())
+		}
+	}
+}
+
+// ConvertData maps the source DB data in vals into Spanner data,
+// based on the Spanner and source DB 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, srcTable string, srcCols []string, vals []string) (string, []string, []interface{}, error) {
+	// Note: if there are many rows for the same srcTable/srcCols,
+	// then the following functionality will be (redundantly)
+	// repeated for every row converted. If this becomes a
+	// performance issue, we could consider moving this block of
+	// code to the callers of ConverData to avoid the redundancy.
+	spTable, err := GetSpannerTable(conv, srcTable)
+	if err != nil {
+		return "", []string{}, []interface{}{}, fmt.Errorf("can't map source table %s", srcTable)
+	}
+	spCols, err := GetSpannerCols(conv, srcTable, srcCols)
+	if err != nil {
+		return "", []string{}, []interface{}{}, fmt.Errorf("can't map source columns %v", srcCols)
+	}
+	spSchema, ok1 := conv.spSchema[spTable]
+	srcSchema, ok2 := conv.srcSchema[srcTable]
 	if !ok1 || !ok2 {
-		return []string{}, []interface{}{}, fmt.Errorf("can't find table %s in schema", spTable)
+		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))
+	if len(spCols) != len(srcCols) || len(spCols) != len(vals) {
+		return "", []string{}, []interface{}{}, fmt.Errorf("ConvertData: spCols, srcCols and vals don't all have the same lengths: len(spCols)=%d, len(srcCols)=%d, len(vals)=%d", len(spCols), len(srcCols), len(vals))
 	}
-	for i, col := range cols {
+	for i, spCol := range spCols {
+		srcCol := srcCols[i]
 		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)
+		spColDef, ok1 := spSchema.ColDefs[spCol]
+		srcColDef, ok2 := srcSchema.ColDefs[srcCol]
+		if !ok1 || !ok2 {
+			return "", []string{}, []interface{}{}, fmt.Errorf("can't find Spanner and source-db schema for col %s", spCol)
 		}
 		var x interface{}
 		var err error
-		if cd.IsArray {
-			x, err = convArray(cd.T, pgCol, conv.location, vals[i])
+		if spColDef.IsArray {
+			x, err = convArray(spColDef.T, srcColDef.Type.Name, conv.location, vals[i])
 		} else {
-			x, err = convScalar(cd.T, pgCol.id, conv.location, vals[i])
+			x, err = convScalar(spColDef.T, srcColDef.Type.Name, conv.location, vals[i])
 		}
 		if err != nil {
-			return []string{}, []interface{}{}, err
+			return "", []string{}, []interface{}{}, err
 		}
 		v = append(v, x)
-		c = append(c, col)
+		c = append(c, spCol)
 	}
 	if aux, ok := conv.syntheticPKeys[spTable]; ok {
 		c = append(c, aux.col)
@@ -73,10 +114,10 @@ func ConvertData(conv *Conv, spTable, pgTable string, cols []string, vals []stri
 		aux.sequence++
 		conv.syntheticPKeys[spTable] = aux
 	}
-	return c, v, nil
+	return spTable, c, v, nil
 }
 
-func convScalar(spannerType ddl.ScalarType, pgType string, location *time.Location, val string) (interface{}, error) {
+func convScalar(spannerType ddl.ScalarType, srcTypeName 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"
@@ -96,7 +137,7 @@ func convScalar(spannerType ddl.ScalarType, pgType string, location *time.Locati
 	case ddl.String:
 		return val, nil
 	case ddl.Timestamp:
-		return convTimestamp(pgType, location, val)
+		return convTimestamp(srcTypeName, location, val)
 	default:
 		return val, fmt.Errorf("data conversion not implemented for type %v", reflect.TypeOf(spannerType))
 	}
@@ -145,7 +186,7 @@ func convInt64(val string) (int64, error) {
 	return i, err
 }
 
-// convTimestamp maps a PostgreSQL timestamp into a go Time (which
+// convTimestamp maps a source DB 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
@@ -153,11 +194,11 @@ func convInt64(val string) (int64, error) {
 // 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) {
+func convTimestamp(srcTypeName 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" {
+	if srcTypeName == "timestamptz" {
 		// PostgreSQL abbreviates timezone to just hour where possible.
 		t, err = time.Parse("2006-01-02 15:04:05Z07", val)
 		if err != nil {
@@ -180,12 +221,12 @@ func convTimestamp(pgType string, location *time.Location, val string) (t time.T
 		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, fmt.Errorf("can't convert to timestamp (posgres type: %s)", srcTypeName)
 	}
 	return t, err
 }
 
-func convArray(spannerType ddl.ScalarType, pgType pgColDef, location *time.Location, v string) (interface{}, error) {
+func convArray(spannerType ddl.ScalarType, srcTypeName string, 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, ...}")
@@ -255,7 +296,7 @@ func convArray(spannerType ddl.ScalarType, pgType pgColDef, location *time.Locat
 	case ddl.Timestamp:
 		var r []time.Time
 		for _, s := range a {
-			t, err := convTimestamp(pgType.id, location, s)
+			t, err := convTimestamp(srcTypeName, location, s)
 			if err != nil {
 				return []time.Time{}, err
 			}
@@ -265,3 +306,14 @@ func convArray(spannerType ddl.ScalarType, pgType pgColDef, location *time.Locat
 	}
 	return []interface{}{}, fmt.Errorf("array type conversion not implemented for type %v", reflect.TypeOf(spannerType))
 }
+
+func byteSize(r *row) int64 {
+	n := int64(len(r.table))
+	for _, c := range r.cols {
+		n += int64(len(c))
+	}
+	for _, v := range r.vals {
+		n += int64(len(v))
+	}
+	return n
+}
diff --git a/internal/postgres/statement.go b/internal/pgdump.go
similarity index 53%
rename from internal/postgres/statement.go
rename to internal/pgdump.go
index 638dc41..96930b5 100644
--- a/internal/postgres/statement.go
+++ b/internal/pgdump.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"fmt"
@@ -20,20 +20,18 @@ import (
 	"strconv"
 	"strings"
 	"time"
-	"unicode"
 
+	pg_query "github.com/lfittl/pg_query_go"
 	nodes "github.com/lfittl/pg_query_go/nodes"
 
-	"github.com/cloudspannerecosystem/harbourbridge/internal"
-	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
 )
 
 type copyOrInsert struct {
-	stmt    stmtType
-	spTable string   // Spanner table name.
-	pgTable string   // Postgres table name.
-	cols    []string // Spanner cols names.
-	vals    []string // Empty for COPY-FROM.
+	stmt  stmtType
+	table string
+	cols  []string
+	vals  []string // Empty for COPY-FROM.
 }
 
 type stmtType int
@@ -43,6 +41,104 @@ const (
 	insert
 )
 
+// ProcessPgDump reads pg_dump data from r and does schema or data conversion,
+// depending on whether conv is configured for schema mode or data mode.
+// In schema mode, ProcessPgDump incrementally builds a schema (updating conv).
+// In data mode, ProcessPgDump uses this schema to convert PostgreSQL data
+// and writes it to Spanner, using the data sink specified in conv.
+func ProcessPgDump(conv *Conv, r *Reader) error {
+	for {
+		startLine := r.LineNumber
+		startOffset := r.Offset
+		b, stmts, err := readAndParseChunk(conv, r)
+		if err != nil {
+			return err
+		}
+		ci := processStatements(conv, stmts)
+		VerbosePrintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) ci=%v\n", startLine, startOffset, len(stmts), r.LineNumber-startLine, len(b), ci != nil)
+		if ci != nil {
+			switch ci.stmt {
+			case copyFrom:
+				processCopyBlock(conv, ci.table, ci.cols, r)
+			case insert:
+				ProcessDataRow(conv, ci.table, ci.cols, ci.vals)
+			}
+		}
+		if r.EOF {
+			break
+		}
+	}
+	if conv.schemaMode() {
+		schemaToDDL(conv)
+		conv.AddPrimaryKeys()
+	}
+
+	return nil
+}
+
+// readAndParseChunk parses a chunk of pg_dump data, returning the bytes read,
+// the parsed AST (nil if nothing read), and whether we've hit end-of-file.
+func readAndParseChunk(conv *Conv, r *Reader) ([]byte, []nodes.Node, error) {
+	var l [][]byte
+	for {
+		b := r.ReadLine()
+		l = append(l, b)
+		// If we see a semicolon or eof, we're likely to have a command, so try to parse it.
+		// Note: we could just parse every iteration, but that would mean more attempts at parsing.
+		if strings.Contains(string(b), ";") || r.EOF {
+			n := 0
+			for i := range l {
+				n += len(l[i])
+			}
+			s := make([]byte, n)
+			n = 0
+			for i := range l {
+				n += copy(s[n:], l[i])
+			}
+			tree, err := pg_query.Parse(string(s))
+			if err == nil {
+				return s, tree.Statements, nil
+			}
+			// Likely causes of failing to parse:
+			// a) complex statements with embedded semicolons e.g. 'CREATE FUNCTION'
+			// b) a semicolon embedded in a multi-line comment, or
+			// c) a semicolon embedded a string constant or column/table name.
+			// We deal with this case by reading another line and trying again.
+			conv.stats.reparsed++
+		}
+		if r.EOF {
+			return nil, nil, fmt.Errorf("Error parsing last %d line(s) of input", len(l))
+		}
+	}
+}
+
+func processCopyBlock(conv *Conv, srcTable string, srcCols []string, r *Reader) {
+	VerbosePrintf("Parsing COPY-FROM stdin block starting at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
+	for {
+		b := r.ReadLine()
+		if string(b) == "\\.\n" || string(b) == "\\.\r\n" {
+			VerbosePrintf("Parsed COPY-FROM stdin block ending at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
+			return
+		}
+		if r.EOF {
+			conv.unexpected("Reached eof while parsing copy-block")
+			return
+		}
+		conv.statsAddRow(srcTable, conv.schemaMode())
+		// We have to read the copy-block data so that we can process the remaining
+		// pg_dump content. However, if we don't want the data, stop here.
+		// In particular, avoid the strings.Split and ProcessDataRow calls below, which
+		// will be expensive for huge datasets.
+		if !conv.dataMode() {
+			continue
+		}
+		// COPY-FROM blocks use tabs to separate data items. Note that space within data
+		// items is significant e.g. if a table row contains data items "a ", " b "
+		// it will be shown in the COPY-FROM block as "a \t b ".
+		ProcessDataRow(conv, srcTable, srcCols, strings.Split(strings.Trim(string(b), "\r\n"), "\t"))
+	}
+}
+
 // processStatements extracts schema information and data from PostgreSQL
 // statements, updating Conv with new schema information, and returning
 // copyOrInsert if a COPY-FROM or INSERT statement is encountered.
@@ -89,24 +185,24 @@ func processAlterTableStmt(conv *Conv, n nodes.AlterTableStmt) {
 		logStmtError(conv, n, fmt.Errorf("relation is nil"))
 		return
 	}
-	pgTable, spTable, err := getTableName(conv, *n.Relation)
+	table, err := getTableName(conv, *n.Relation)
 	if err != nil {
 		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
 		return
 	}
-	if _, ok := conv.spSchema[spTable]; ok {
+	if _, ok := conv.srcSchema[table]; ok {
 		for _, i := range n.Cmds.Items {
 			switch a := i.(type) {
 			case nodes.AlterTableCmd:
 				switch {
 				case a.Subtype == nodes.AT_SetNotNull && a.Name != nil:
-					c := constraint{ct: nodes.CONSTR_NOTNULL, keys: []string{*a.Name}}
-					updateSchema(conv, spTable, pgTable, []constraint{c}, "ALTER TABLE")
+					c := constraint{ct: nodes.CONSTR_NOTNULL, cols: []string{*a.Name}}
+					updateSchema(conv, table, []constraint{c}, "ALTER TABLE")
 					conv.schemaStatement([]nodes.Node{n, a})
 				case a.Subtype == nodes.AT_AddConstraint && a.Def != nil:
 					switch d := a.Def.(type) {
 					case nodes.Constraint:
-						updateSchema(conv, spTable, pgTable, extractConstraints(conv, n, pgTable, []nodes.Node{d}), "ALTER TABLE")
+						updateSchema(conv, table, extractConstraints(conv, n, table, []nodes.Node{d}), "ALTER TABLE")
 						conv.schemaStatement([]nodes.Node{n, a, d})
 					default:
 						conv.skipStatement([]nodes.Node{n, a, d})
@@ -125,19 +221,18 @@ func processAlterTableStmt(conv *Conv, n nodes.AlterTableStmt) {
 		// For debugging purposes we log the lookup failure if we're
 		// in verbose mode, but otherwise  we just skip these statements.
 		conv.skipStatement([]nodes.Node{n})
-		internal.VerbosePrintf("Processing %v statement: table %s not found", reflect.TypeOf(n), spTable)
+		VerbosePrintf("Processing %v statement: table %s not found", reflect.TypeOf(n), table)
 	}
 }
 
 func processCreateStmt(conv *Conv, n nodes.CreateStmt) {
-	var cols []string
-	cds := make(map[string]ddl.ColumnDef)
-	pgCols := make(map[string]pgColDef)
+	var colNames []string
+	colDef := make(map[string]schema.Column)
 	if n.Relation == nil {
 		logStmtError(conv, n, fmt.Errorf("relation is nil"))
 		return
 	}
-	pgTable, spTable, err := getTableName(conv, *n.Relation)
+	table, err := getTableName(conv, *n.Relation)
 	if err != nil {
 		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
 		return
@@ -146,67 +241,47 @@ func processCreateStmt(conv *Conv, n nodes.CreateStmt) {
 	for _, te := range n.TableElts.Items {
 		switch i := te.(type) {
 		case nodes.ColumnDef:
-			col, cd, pgCol, cdConstraints, err := processColumnDef(conv, i, pgTable)
+			name, col, cdConstraints, err := processColumn(conv, i, table)
 			if err != nil {
 				logStmtError(conv, n, err)
 				return
 			}
-			cols = append(cols, col)
-			cds[col] = cd
-			pgCols[col] = pgCol
+			colNames = append(colNames, name)
+			colDef[name] = col
 			constraints = append(constraints, cdConstraints...)
 		case nodes.Constraint:
-			// Note: there should be at most one Constraint nodes in
+			// Note: there should be at most one Constraint node in
 			// n.TableElts.Items. We don't check this. We just keep
 			// collecting constraints.
-			constraints = extractConstraints(conv, n, pgTable, []nodes.Node{i})
+			constraints = extractConstraints(conv, n, table, []nodes.Node{i})
 		default:
 			conv.unexpected(fmt.Sprintf("Found %s node while processing CreateStmt TableElts", prNodeType(i)))
 		}
 	}
 	conv.schemaStatement([]nodes.Node{n})
-	conv.spSchema[spTable] = ddl.CreateTable{spTable, cols, cds, nil, mkTableComment(pgTable)}
-	conv.pgSchema[pgTable] = pgTableDef{pgCols}
+	conv.srcSchema[table] = schema.Table{
+		Name:     table,
+		ColNames: colNames,
+		ColDefs:  colDef}
 	// Note: constraints contains all info about primary keys and not-null keys.
-	updateSchema(conv, spTable, pgTable, constraints, "CREATE TABLE")
+	updateSchema(conv, table, constraints, "CREATE TABLE")
 }
 
-func processColumnDef(conv *Conv, n nodes.ColumnDef, pgTable string) (string, ddl.ColumnDef, pgColDef, []constraint, error) {
+func processColumn(conv *Conv, n nodes.ColumnDef, table string) (string, schema.Column, []constraint, error) {
 	mods := getTypeMods(conv, n.TypeName.Typmods)
 	if n.Colname == nil {
-		return "", ddl.ColumnDef{}, pgColDef{}, nil, fmt.Errorf("colname is nil")
-	}
-	pgCol := *n.Colname
-	spCol, err := GetSpannerCol(conv, pgTable, pgCol, false)
-	if err != nil {
-		return "", ddl.ColumnDef{}, pgColDef{}, nil, fmt.Errorf("can't get Spanner col: %w", err)
+		return "", schema.Column{}, nil, fmt.Errorf("colname is nil")
 	}
+	name := *n.Colname
 	tid, err := getTypeID(n.TypeName.Names.Items)
 	if err != nil {
-		return "", ddl.ColumnDef{}, pgColDef{}, nil, fmt.Errorf("can't get type id for %s: %w", spCol, err)
-	}
-	t, issues := getSpannerType(conv, tid, mods)
-
-	// Treatment of arrays: if the PostgreSQL type has a single array parameter
-	// we map it to a Spanner array. The array bounds themselves are ignored
-	// since Spanner doesn't support them (and PostgreSQL ignores them anyway:
-	// https://www.postgresql.org/docs/9.1/arrays.html).
-	// Spanner does not support multi-dimensional arrays -- we map those
-	// to STRING(MAX).
-	a := getArrayBounds(conv, n.TypeName.ArrayBounds)
-	if len(a) > 1 {
-		t = ddl.String{ddl.MaxLength{}}
-		issues = append(issues, multiDimensionalArray)
-	}
-	constraints := analyzeColDefConstraints(conv, n, pgTable, n.Constraints.Items, pgCol)
-	pgSchema := pgColDef{id: tid, mods: mods, array: a, issues: issues}
-	cd := ddl.ColumnDef{
-		Name:    spCol,
-		T:       t,
-		IsArray: len(a) == 1,
-		Comment: mkColComment(pgCol, printType(pgSchema)),
-	}
-	return spCol, cd, pgSchema, constraints, nil
+		return "", schema.Column{}, nil, fmt.Errorf("can't get type id for %s: %w", name, err)
+	}
+	ty := schema.Type{
+		Name:        tid,
+		Mods:        mods,
+		ArrayBounds: getArrayBounds(conv, n.TypeName.ArrayBounds)}
+	return name, schema.Column{Name: name, Type: ty}, analyzeColDefConstraints(conv, n, table, n.Constraints.Items, name), nil
 }
 
 func processInsertStmt(conv *Conv, n nodes.InsertStmt) *copyOrInsert {
@@ -214,16 +289,16 @@ func processInsertStmt(conv *Conv, n nodes.InsertStmt) *copyOrInsert {
 		logStmtError(conv, n, fmt.Errorf("relation is nil"))
 		return nil
 	}
-	pgTable, spTable, err := getTableName(conv, *n.Relation)
+	table, err := getTableName(conv, *n.Relation)
 	if err != nil {
 		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
 		return nil
 	}
-	conv.statsAddRow(spTable, conv.schemaMode())
-	cols, err := getCols(conv, pgTable, n.Cols.Items)
+	conv.statsAddRow(table, conv.schemaMode())
+	colNames, err := getCols(conv, table, n.Cols.Items)
 	if err != nil {
 		logStmtError(conv, n, fmt.Errorf("can't get col name: %w", err))
-		conv.statsAddBadRow(spTable, conv.schemaMode())
+		conv.statsAddBadRow(table, conv.schemaMode())
 		return nil
 	}
 	var values []string
@@ -232,7 +307,7 @@ func processInsertStmt(conv *Conv, n nodes.InsertStmt) *copyOrInsert {
 		values = getVals(conv, sel.ValuesLists, n)
 		conv.dataStatement([]nodes.Node{n})
 		if conv.dataMode() {
-			return &copyOrInsert{stmt: insert, spTable: spTable, pgTable: pgTable, cols: cols, vals: values}
+			return &copyOrInsert{stmt: insert, table: table, cols: colNames, vals: values}
 		}
 	default:
 		conv.unexpected(fmt.Sprintf("Found %s node while processing InsertStmt SelectStmt", prNodeType(sel)))
@@ -245,11 +320,10 @@ func processCopyStmt(conv *Conv, n nodes.CopyStmt) *copyOrInsert {
 	// encounter errors. Otherwise we won't be able to parse
 	// the data portion of the COPY-FROM statement, and we'll
 	// likely get stuck at this point in the pg_dump file.
-	pgTable := "BOGUS_COPY_FROM_TABLE"
-	spTable := "BOGUS_COPY_FROM_TABLE"
+	table := "BOGUS_COPY_FROM_TABLE"
 	var err error
 	if n.Relation != nil {
-		pgTable, spTable, err = getTableName(conv, *n.Relation)
+		table, err = getTableName(conv, *n.Relation)
 		if err != nil {
 			conv.unexpected(fmt.Sprintf("Processing %v statement: %s", reflect.TypeOf(n), err))
 		}
@@ -259,9 +333,6 @@ func processCopyStmt(conv *Conv, n nodes.CopyStmt) *copyOrInsert {
 	var cols []string
 	for _, a := range n.Attlist.Items {
 		s, err := getString(a)
-		if err == nil {
-			s, err = GetSpannerCol(conv, pgTable, s, true)
-		}
 		if err != nil {
 			conv.unexpected(fmt.Sprintf("Processing %v statement Attlist: %s", reflect.TypeOf(n), err))
 			s = "BOGUS_COPY_FROM_COLUMN"
@@ -269,7 +340,7 @@ func processCopyStmt(conv *Conv, n nodes.CopyStmt) *copyOrInsert {
 		cols = append(cols, s)
 	}
 	conv.dataStatement([]nodes.Node{n})
-	return &copyOrInsert{stmt: copyFrom, spTable: spTable, pgTable: pgTable, cols: cols}
+	return &copyOrInsert{stmt: copyFrom, table: table, cols: cols}
 }
 
 func processVariableSetStmt(conv *Conv, n nodes.VariableSetStmt) {
@@ -296,83 +367,6 @@ func processVariableSetStmt(conv *Conv, n nodes.VariableSetStmt) {
 	}
 }
 
-// getSpannerType determines the Spanner type for the scalar postgres type
-// defined by id and mods. This is the core postgres-to-Spanner type mapping.
-// getSpannerType returns the Spanner type and the schema issues encountered
-// during conversion.
-func getSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []schemaIssue) {
-	maxExpectedMods := func(n int) {
-		if len(mods) > n {
-			conv.unexpected(fmt.Sprintf("Found %d mods while processing type id=%s", len(mods), id))
-		}
-	}
-	switch id {
-	case "bool":
-		maxExpectedMods(0)
-		return ddl.Bool{}, nil
-	case "bigserial":
-		maxExpectedMods(0)
-		return ddl.Int64{}, []schemaIssue{serial}
-	case "bpchar": // Note: Postgres internal name for char is bpchar (aka blank padded char).
-		maxExpectedMods(1)
-		if len(mods) > 0 {
-			return ddl.String{ddl.Int64Length{mods[0]}}, nil
-		}
-		// Note: bpchar without length specifier is equivalent to bpchar(1)
-		return ddl.String{ddl.Int64Length{1}}, nil
-	case "bytea":
-		maxExpectedMods(0)
-		return ddl.Bytes{ddl.MaxLength{}}, nil
-	case "date":
-		maxExpectedMods(0)
-		return ddl.Date{}, nil
-	case "float8":
-		maxExpectedMods(0)
-		return ddl.Float64{}, nil
-	case "float4":
-		maxExpectedMods(0)
-		return ddl.Float64{}, []schemaIssue{widened}
-	case "int8":
-		maxExpectedMods(0)
-		return ddl.Int64{}, nil
-	case "int4":
-		maxExpectedMods(0)
-		return ddl.Int64{}, []schemaIssue{widened}
-	case "int2":
-		maxExpectedMods(0)
-		return ddl.Int64{}, []schemaIssue{widened}
-	case "numeric": // Map all numeric types to float64.
-		maxExpectedMods(2)
-		if len(mods) > 0 && mods[0] <= 15 {
-			// float64 can represent this numeric type faithfully.
-			// Note: int64 has 53 bits for mantissa, which is ~15.96
-			// decimal digits.
-			return ddl.Float64{}, []schemaIssue{numericThatFits}
-		}
-		return ddl.Float64{}, []schemaIssue{numeric}
-	case "serial":
-		maxExpectedMods(0)
-		return ddl.Int64{}, []schemaIssue{serial}
-	case "text":
-		maxExpectedMods(0)
-		return ddl.String{ddl.MaxLength{}}, nil
-	case "timestamptz":
-		maxExpectedMods(1)
-		return ddl.Timestamp{}, nil
-	case "timestamp":
-		maxExpectedMods(1)
-		// Map timestamp without timezone to Spanner timestamp.
-		return ddl.Timestamp{}, []schemaIssue{timestamp}
-	case "varchar":
-		maxExpectedMods(1)
-		if len(mods) > 0 {
-			return ddl.String{ddl.Int64Length{mods[0]}}, nil
-		}
-		return ddl.String{ddl.MaxLength{}}, nil
-	}
-	return ddl.String{ddl.MaxLength{}}, []schemaIssue{noGoodType}
-}
-
 func getTypeMods(conv *Conv, t nodes.List) (l []int64) {
 	for _, x := range t.Items {
 		switch t1 := x.(type) {
@@ -424,33 +418,9 @@ func getTypeID(l []nodes.Node) (string, error) {
 	return strings.Join(ids, "."), nil
 }
 
-func printType(pgCol pgColDef) string {
-	s := pgCol.id
-	if len(pgCol.mods) > 0 {
-		var l []string
-		for _, x := range pgCol.mods {
-			l = append(l, strconv.FormatInt(x, 10))
-		}
-		s = fmt.Sprintf("%s(%s)", s, strings.Join(l, ","))
-	}
-	if len(pgCol.array) > 0 {
-		l := []string{s}
-		for _, x := range pgCol.array {
-			if x == -1 {
-				l = append(l, "[]")
-			} else {
-				l = append(l, fmt.Sprintf("[%d]", x))
-			}
-		}
-		s = strings.Join(l, "")
-	}
-	return s
-}
-
 // getTableName extracts the table name from RangeVar n, and returns
-// the raw extracted name (the PostgreSQL table name) plus the corresponding
-// Spanner table name.
-func getTableName(conv *Conv, n nodes.RangeVar) (string, string, error) {
+// the raw extracted name (the PostgreSQL table name).
+func getTableName(conv *Conv, n nodes.RangeVar) (string, error) {
 	// RangeVar is used to represent table names. It consists of three components:
 	//  Catalogname: database name; either not specified or the current database
 	//  Schemaname: schemas are PostgreSql namepaces; often unspecified; defaults to "public"
@@ -470,41 +440,36 @@ func getTableName(conv *Conv, n nodes.RangeVar) (string, string, error) {
 		l = append(l, *n.Schemaname)
 	}
 	if n.Relname == nil && *n.Relname == "" {
-		return "", "", fmt.Errorf("relname is empty: can't build table name")
+		return "", fmt.Errorf("relname is empty: can't build table name")
 	}
 	l = append(l, *n.Relname)
-	pgTable := strings.Join(l, ".")
-	spTable, err := GetSpannerTable(conv, pgTable)
-	if err != nil {
-		return "", "", err
-	}
-	return pgTable, spTable, nil
+	return strings.Join(l, "."), nil
 }
 
 type constraint struct {
 	ct   nodes.ConstrType
-	keys []string
+	cols []string
 }
 
 // extractConstraints traverses a list of nodes (expecting them to be
 // Constraint nodes), and collects the contraints they represent as
-// a list of constraint-type/key-list pairs.
-func extractConstraints(conv *Conv, n nodes.Node, pgTable string, l []nodes.Node) (cs []constraint) {
+// a list of constraint-type/column-names pairs.
+func extractConstraints(conv *Conv, n nodes.Node, table string, l []nodes.Node) (cs []constraint) {
 	for _, i := range l {
 		switch d := i.(type) {
 		case nodes.Constraint:
-			var keys []string
+			var cols []string
 			for _, j := range d.Keys.Items {
 				k, err := getString(j)
 				if err == nil {
-					keys = append(keys, k)
+					cols = append(cols, k)
 				}
 				if err != nil {
 					conv.unexpected(fmt.Sprintf("Processing %v statement: error processing constraints: %s", reflect.TypeOf(n), err.Error()))
 					conv.errorInStatement([]nodes.Node{n, d})
 				}
 			}
-			cs = append(cs, constraint{d.Contype, keys})
+			cs = append(cs, constraint{ct: d.Contype, cols: cols})
 		default:
 			conv.unexpected(fmt.Sprintf("Processing %v statement: found %s node while processing constraints\n", reflect.TypeOf(n), reflect.TypeOf(d)))
 		}
@@ -515,95 +480,79 @@ func extractConstraints(conv *Conv, n nodes.Node, pgTable string, l []nodes.Node
 // analyzeColDefConstraints is like extractConstraints, but is specifially for
 // ColDef constraints. These constraints don't specify a key since they
 // are constraints for the column defined by ColDef.
-func analyzeColDefConstraints(conv *Conv, n nodes.Node, pgTable string, l []nodes.Node, pgCol string) (cs []constraint) {
+func analyzeColDefConstraints(conv *Conv, n nodes.Node, table string, l []nodes.Node, pgCol string) (cs []constraint) {
 	// Do generic constraint processing and then set the keys of each constraint
 	// to {pgCol}.
-	for _, c := range extractConstraints(conv, n, pgTable, l) {
-		if len(c.keys) != 0 {
+	for _, c := range extractConstraints(conv, n, table, l) {
+		if len(c.cols) != 0 {
 			conv.unexpected("ColumnDef constraint has keys")
 		}
-		c.keys = []string{pgCol}
+		c.cols = []string{pgCol}
 		cs = append(cs, c)
 	}
 	return cs
 }
 
-// getConstraint finds the constraint-type ct in the constraint list cs.
-// It returns the keys from the constraint (if found) and a boolean to
-// indicate if the constraint was found or not.
-func getConstraint(ct nodes.ConstrType, cs []constraint) ([]string, bool) {
+// updateSchema updates the schema for table based on the given constraints.
+// 's' is the statement type being processed, and is used for debug messages.
+func updateSchema(conv *Conv, table string, cs []constraint, s string) {
 	for _, c := range cs {
-		if c.ct == ct {
-			return c.keys, true
+		switch c.ct {
+		case nodes.CONSTR_PRIMARY:
+			ct := conv.srcSchema[table]
+			checkEmpty(conv, ct.PrimaryKeys, s)
+			ct.PrimaryKeys = toSchemaKeys(conv, table, c.cols) // Drop any previous primary keys.
+			// In Spanner, primary key columns are usually annotated with NOT NULL,
+			// but this can be omitted to allow NULL values in key columns.
+			// In PostgreSQL, the primary key constraint is a combination of
+			// NOT NULL and UNIQUE i.e. primary keys must be NOT NULL.
+			// We preserve PostgreSQL semantics and enforce NOT NULL.
+			updateCols(nodes.CONSTR_NOTNULL, c.cols, ct.ColDefs)
+			conv.srcSchema[table] = ct
+		default:
+			ct := conv.srcSchema[table]
+			updateCols(c.ct, c.cols, ct.ColDefs)
+			conv.srcSchema[table] = ct
 		}
 	}
-	return []string{}, false
 }
 
-// updateSchema updates the schema for spTable based on the given constraints.
-// 's' is the statement type being processed, and is used for debug messages.
-func updateSchema(conv *Conv, spTable, pgTable string, cs []constraint, s string) {
-	if len(cs) == 0 {
-		return
-	}
-	if keys, found := getConstraint(nodes.CONSTR_PRIMARY, cs); found {
-		ct := conv.spSchema[spTable]
-		checkEmpty(conv, ct.Pks, s)
-		ct.Pks = toDdlPkeys(conv, pgTable, keys) // Drop any previous primary keys.
-		// In Spanner, primary key columns are usually annotated with NOT NULL,
-		// but this can be omitted to allow NULL values in key columns.
-		// In PostgreSQL, the primary key constraint is a combination of
-		// NOT NULL and UNIQUE i.e. primary keys must be NOT NULL.
-		// We preserve PostgreSQL semantics and enforce NOT NULL.
-		ct.Cds = setNotNull(ct.Pks, ct.Cds)
-		conv.spSchema[spTable] = ct
-	}
-	if keys, found := getConstraint(nodes.CONSTR_NOTNULL, cs); found {
-		ct := conv.spSchema[spTable]
-		ct.Cds = setNotNull(toDdlPkeys(conv, pgTable, keys), ct.Cds)
-		conv.spSchema[spTable] = ct
-	}
-	for _, c := range cs {
-		for _, col := range c.keys {
-			pgCol := conv.pgSchema[pgTable].cols[col]
-			switch c.ct {
-			case nodes.CONSTR_DEFAULT:
-				pgCol.issues = append(pgCol.issues, defaultValue)
-			case nodes.CONSTR_FOREIGN:
-				pgCol.issues = append(pgCol.issues, foreignKey)
-			}
-			conv.pgSchema[pgTable].cols[col] = pgCol
+// updateCols updates colDef with new constraints. Specifically, we apply
+// 'ct' to each column in colNames.
+func updateCols(ct nodes.ConstrType, colNames []string, colDef map[string]schema.Column) {
+	// TODO: add cases for other constraints.
+	for _, c := range colNames {
+		cd := colDef[c]
+		switch ct {
+		case nodes.CONSTR_NOTNULL:
+			cd.NotNull = true
+		case nodes.CONSTR_DEFAULT:
+			cd.Ignored.Default = true
+		case nodes.CONSTR_FOREIGN:
+			cd.Ignored.ForeignKey = true
 		}
+		colDef[c] = cd
 	}
 }
 
-// toDdlPkeys converts a string list of PostgreSQL primary keys to
-// Spanner ddl primary keys.
-func toDdlPkeys(conv *Conv, pgTable string, s []string) (l []ddl.IndexKey) {
+// toSchemaKeys converts a string list of PostgreSQL primary keys to
+// schema primary keys.
+func toSchemaKeys(conv *Conv, table string, s []string) (l []schema.Key) {
 	for _, k := range s {
-		col, err := GetSpannerCol(conv, pgTable, k, true)
-		if err == nil {
-			// PostgreSQL primary keys have no notation of ascending/descending.
-			// We map them all into Spanner ascending primarary keys (the default).
-			l = append(l, ddl.IndexKey{Col: col})
-		} else {
-			conv.unexpected(fmt.Sprintf("Can't get Spanner col: %s", err.Error()))
-		}
+		// PostgreSQL primary keys have no notation of ascending/descending.
+		// We map them all into ascending primarary keys.
+		l = append(l, schema.Key{Column: k})
 	}
 	return l
 }
 
-// getCols extracts and returns the columns for an InsertStatement.
-func getCols(conv *Conv, pgTable string, l []nodes.Node) (cols []string, err error) {
+// getCols extracts and returns the column names for an InsertStatement.
+func getCols(conv *Conv, table string, l []nodes.Node) (cols []string, err error) {
 	for _, n := range l {
 		switch r := n.(type) {
 		case nodes.ResTarget:
 			if r.Name != nil {
-				col, err := GetSpannerCol(conv, pgTable, *r.Name, true)
-				if err != nil {
-					return nil, fmt.Errorf("can't get col name: %w", err)
-				}
-				cols = append(cols, col)
+				cols = append(cols, *r.Name)
 			}
 		default:
 			return nil, fmt.Errorf("expecting ResTarget node but got %v node while processing Cols", reflect.TypeOf(r))
@@ -654,43 +603,8 @@ func getString(node nodes.Node) (string, error) {
 
 // checkEmpty verifies that pkeys is empty and generates a warning if it isn't.
 // PostgreSQL explicitly forbids multiple primary keys.
-func checkEmpty(conv *Conv, pkeys []ddl.IndexKey, s string) {
+func checkEmpty(conv *Conv, pkeys []schema.Key, s string) {
 	if len(pkeys) != 0 {
 		conv.unexpected(fmt.Sprintf("%s statement is adding a second primary key", s))
 	}
 }
-
-func mkColComment(name, t string) string {
-	if needQuote(name) {
-		name = strconv.Quote(name)
-	}
-	return "From PostgreSQL: " + name + " " + t
-}
-
-func mkTableComment(name string) string {
-	if needQuote(name) {
-		name = strconv.Quote(name)
-	}
-	return "Spanner schema for PostgreSQL table " + name
-}
-
-// needQuote returns true if we need to quote this name before
-// printing it as part of schema comments.
-func needQuote(s string) bool {
-	for _, r := range s {
-		if unicode.IsLetter(r) || unicode.IsDigit(r) || unicode.IsPunct(r) {
-			continue
-		}
-		return true
-	}
-	return false
-}
-
-func setNotNull(pkeys []ddl.IndexKey, cds map[string]ddl.ColumnDef) map[string]ddl.ColumnDef {
-	for _, pk := range pkeys {
-		cd := cds[pk.Col]
-		cd.NotNull = true
-		cds[pk.Col] = cd
-	}
-	return cds
-}
diff --git a/internal/postgres/process.go b/internal/postgres/process.go
deleted file mode 100644
index 586e230..0000000
--- a/internal/postgres/process.go
+++ /dev/null
@@ -1,165 +0,0 @@
-// 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 implements PostgreSQL-specific schema conversion
-// and data conversion for HarbourBridge.
-package postgres
-
-import (
-	"fmt"
-	"strings"
-
-	pg_query "github.com/lfittl/pg_query_go"
-	nodes "github.com/lfittl/pg_query_go/nodes"
-
-	"github.com/cloudspannerecosystem/harbourbridge/internal"
-)
-
-// ProcessPgDump reads pg_dump data from r and does schema or data conversion,
-// depending on whether conv is configured for schema mode or data mode.
-// In schema mode, ProcessPgDump incrementally builds a schema (updating conv).
-// In data mode, ProcessPgDump uses this schema to convert PostgreSQL data
-// and writes it to Spanner, using the data sink specified in conv.
-func ProcessPgDump(conv *Conv, r *internal.Reader) error {
-	for {
-		startLine := r.LineNumber
-		startOffset := r.Offset
-		b, stmts, err := readAndParseChunk(conv, r)
-		if err != nil {
-			return err
-		}
-		ci := processStatements(conv, stmts)
-		internal.VerbosePrintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) ci=%v\n", startLine, startOffset, len(stmts), r.LineNumber-startLine, len(b), ci != nil)
-		if ci != nil {
-			switch ci.stmt {
-			case copyFrom:
-				processCopyBlock(conv, ci, r)
-			case insert:
-				ProcessRow(conv, ci.spTable, ci.pgTable, ci.cols, ci.vals)
-			}
-		}
-		if r.EOF {
-			break
-		}
-	}
-	if conv.schemaMode() {
-		conv.AddPrimaryKeys()
-	}
-
-	return nil
-}
-
-// ProcessRow converts a row of data and writes it out to Spanner.
-// spTable and pgTable are the Spanner and PostgreSQL table names respectively
-// (typically they are the same), cols are Spanner cols, and vals contains
-// string data to be converted to appropriate types to send to Spanner.
-// ProcessRow is only called in dataMode.
-func ProcessRow(conv *Conv, spTable, pgTable string, cols, vals []string) {
-	c, v, err := ConvertData(conv, spTable, pgTable, cols, vals)
-	if err != nil {
-		conv.unexpected(fmt.Sprintf("Error while converting data: %s\n", err))
-		conv.statsAddBadRow(spTable, conv.dataMode())
-		r := &row{table: spTable, cols: cols, 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
-		}
-	} else {
-		if conv.dataSink == nil {
-			msg := "Internal error: ProcessRow called but dataSink not configured"
-			internal.VerbosePrintf("%s\n", msg)
-			conv.unexpected(msg)
-			conv.statsAddBadRow(spTable, conv.dataMode())
-		} else {
-			conv.dataSink(spTable, c, v)
-			conv.statsAddGoodRow(spTable, conv.dataMode())
-		}
-	}
-}
-
-func byteSize(r *row) int64 {
-	n := int64(len(r.table))
-	for _, c := range r.cols {
-		n += int64(len(c))
-	}
-	for _, v := range r.vals {
-		n += int64(len(v))
-	}
-	return n
-}
-
-// readAndParseChunk parses a chunk of pg_dump data, returning the bytes read,
-// the parsed AST (nil if nothing read), and whether we've hit end-of-file.
-func readAndParseChunk(conv *Conv, r *internal.Reader) ([]byte, []nodes.Node, error) {
-	var l [][]byte
-	for {
-		b := r.ReadLine()
-		l = append(l, b)
-		// If we see a semicolon or eof, we're likely to have a command, so try to parse it.
-		// Note: we could just parse every iteration, but that would mean more attempts at parsing.
-		if strings.Contains(string(b), ";") || r.EOF {
-			n := 0
-			for i := range l {
-				n += len(l[i])
-			}
-			s := make([]byte, n)
-			n = 0
-			for i := range l {
-				n += copy(s[n:], l[i])
-			}
-			tree, err := pg_query.Parse(string(s))
-			if err == nil {
-				return s, tree.Statements, nil
-			}
-			// Likely causes of failing to parse:
-			// a) complex statements with embedded semicolons e.g. 'CREATE FUNCTION'
-			// b) a semicolon embedded in a multi-line comment, or
-			// c) a semicolon embedded a string constant or column/table name.
-			// We deal with this case by reading another line and trying again.
-			conv.stats.reparsed++
-		}
-		if r.EOF {
-			return nil, nil, fmt.Errorf("Error parsing last %d line(s) of input", len(l))
-		}
-	}
-}
-
-func processCopyBlock(conv *Conv, c *copyOrInsert, r *internal.Reader) {
-	internal.VerbosePrintf("Parsing COPY-FROM stdin block starting at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
-	for {
-		b := r.ReadLine()
-		if string(b) == "\\.\n" || string(b) == "\\.\r\n" {
-			internal.VerbosePrintf("Parsed COPY-FROM stdin block ending at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
-			return
-		}
-		if r.EOF {
-			conv.unexpected("Reached eof while parsing copy-block")
-			return
-		}
-		conv.statsAddRow(c.spTable, conv.schemaMode())
-		// We have to read the copy-block data so that we can process the remaining
-		// pg_dump content. However, if we don't want the data, stop here.
-		// In particular, avoid the strings.Split and ProcessRow calls below, which
-		// will be expensive for huge datasets.
-		if !conv.dataMode() {
-			continue
-		}
-		// COPY-FROM blocks use tabs to separate data items. Note that space within data
-		// items is significant e.g. if a table row contains data items "a ", " b "
-		// it will be shown in the COPY-FROM block as "a \t b ".
-		ProcessRow(conv, c.spTable, c.pgTable, c.cols, strings.Split(strings.Trim(string(b), "\r\n"), "\t"))
-	}
-}
diff --git a/internal/postgres/report.go b/internal/report.go
similarity index 83%
rename from internal/postgres/report.go
rename to internal/report.go
index 84f66f3..6da0398 100644
--- a/internal/postgres/report.go
+++ b/internal/report.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"bufio"
@@ -20,6 +20,7 @@ import (
 	"sort"
 	"strings"
 
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
@@ -33,7 +34,7 @@ func GenerateReport(conv *Conv, w *bufio.Writer, badWrites map[string]int64) str
 	ignored := ignoredStatements(conv)
 	w.WriteString("\n")
 	if len(ignored) > 0 {
-		justifyLines(w, fmt.Sprintf("Note that the following PostgreSQL statements "+
+		justifyLines(w, fmt.Sprintf("Note that the following source DB statements "+
 			"were detected but ignored: %s.",
 			strings.Join(ignored, ", ")), 80, 0)
 		w.WriteString("\n\n")
@@ -47,8 +48,8 @@ func GenerateReport(conv *Conv, w *bufio.Writer, badWrites map[string]int64) str
 	w.WriteString("\n\n")
 	writeStmtStats(conv, w)
 	for _, t := range reports {
-		h := fmt.Sprintf("Table %s", t.pgTable)
-		if t.pgTable != t.spTable {
+		h := fmt.Sprintf("Table %s", t.srcTable)
+		if t.srcTable != t.spTable {
 			h = h + fmt.Sprintf(" (mapped to Spanner table %s)", t.spTable)
 		}
 		writeHeading(w, h)
@@ -67,7 +68,7 @@ func GenerateReport(conv *Conv, w *bufio.Writer, badWrites map[string]int64) str
 }
 
 type tableReport struct {
-	pgTable       string
+	srcTable      string
 	spTable       string
 	rows          int64
 	badRows       int64
@@ -86,41 +87,41 @@ func analyzeTables(conv *Conv, badWrites map[string]int64) (r []tableReport) {
 	// Process tables in alphabetical order. This ensures that tables
 	// appear in alphabetical order in report.txt.
 	var tables []string
-	for t := range conv.pgSchema {
+	for t := range conv.srcSchema {
 		tables = append(tables, t)
 	}
 	sort.Strings(tables)
-	for _, pgTable := range tables {
-		r = append(r, buildTableReport(conv, pgTable, badWrites))
+	for _, srcTable := range tables {
+		r = append(r, buildTableReport(conv, srcTable, badWrites))
 	}
 	return r
 }
 
-func buildTableReport(conv *Conv, pgTable string, badWrites map[string]int64) tableReport {
-	spTable, err := GetSpannerTable(conv, pgTable)
-	pgSchema, ok1 := conv.pgSchema[pgTable]
+func buildTableReport(conv *Conv, srcTable string, badWrites map[string]int64) tableReport {
+	spTable, err := GetSpannerTable(conv, srcTable)
+	srcSchema, ok1 := conv.srcSchema[srcTable]
 	spSchema, ok2 := conv.spSchema[spTable]
-	tr := tableReport{pgTable: pgTable, spTable: spTable}
+	tr := tableReport{srcTable: srcTable, spTable: spTable}
 	if err != nil || !ok1 || !ok2 {
-		m := "bad PostgreSQL-to-Spanner table mapping or Spanner schema"
+		m := "bad source-DB-to-Spanner table mapping or Spanner schema"
 		conv.unexpected("report: " + m)
 		tr.body = []tableReportBody{tableReportBody{heading: "Internal error: " + m}}
 		return tr
 	}
-	issues, cols, warnings := analyzeCols(conv, pgTable, spTable)
+	issues, cols, warnings := analyzeCols(conv, srcTable, spTable)
 	tr.cols = cols
 	tr.warnings = warnings
 	if pk, ok := conv.syntheticPKeys[spTable]; ok {
 		tr.syntheticPKey = pk.col
-		tr.body = buildTableReportBody(conv, pgTable, issues, spSchema, pgSchema, &pk.col)
+		tr.body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, &pk.col)
 	} else {
-		tr.body = buildTableReportBody(conv, pgTable, issues, spSchema, pgSchema, nil)
+		tr.body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil)
 	}
-	fillRowStats(conv, spTable, badWrites, &tr)
+	fillRowStats(conv, srcTable, badWrites, &tr)
 	return tr
 }
 
-func buildTableReportBody(conv *Conv, pgTable string, issues map[string][]schemaIssue, spSchema ddl.CreateTable, pgSchema pgTableDef, syntheticPK *string) []tableReportBody {
+func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]schemaIssue, spSchema ddl.CreateTable, srcSchema schema.Table, syntheticPK *string) []tableReportBody {
 	var body []tableReportBody
 	for _, p := range []struct {
 		heading  string
@@ -138,15 +139,15 @@ func buildTableReportBody(conv *Conv, pgTable string, issues map[string][]schema
 		var l []string
 		if syntheticPK != nil {
 			// Warnings about synthetic primary keys must be handled as a special case
-			// because we have a Spanner column with no matching PostgreSQL col.
+			// because we have a Spanner column with no matching source DB col.
 			// Much of the generic code for processing issues assumes we have both.
 			if p.severity == warning {
 				l = append(l, fmt.Sprintf("Column '%s' was added because this table didn't have a primary key. Spanner requires a primary key for every table", *syntheticPK))
 			}
 		}
 		issueBatcher := make(map[schemaIssue]bool)
-		for _, pgCol := range cols {
-			for _, i := range issues[pgCol] {
+		for _, srcCol := range cols {
+			for _, i := range issues[srcCol] {
 				if issueDB[i].severity != p.severity {
 					continue
 				}
@@ -158,32 +159,33 @@ func buildTableReportBody(conv *Conv, pgTable string, issues map[string][]schema
 					}
 					issueBatcher[i] = true
 				}
-				spCol, err := GetSpannerCol(conv, pgTable, pgCol, true)
+				spCol, err := GetSpannerCol(conv, srcTable, srcCol, true)
 				if err != nil {
 					conv.unexpected(err.Error())
 				}
-				pgType := printType(pgSchema.cols[pgCol])
-				sType := spSchema.Cds[spCol].PrintColumnDefType()
-				// A note on case: Both PostgreSQL types and Spanner types are case
-				// insensitive. PostgreSQL often uses lower case (e.g. pg_dump uses
-				// lower case) and pg_query uses lower case, so pgType is lower-case.
-				// However Spanner defaults to upper case, and the Spanner AST uses
-				// upper case, so sType is upper case. When printing PostgreSQL and
+				srcType := printSourceType(srcSchema.ColDefs[srcCol].Type)
+				spType := spSchema.ColDefs[spCol].PrintColumnDefType()
+				// A note on case: Spanner types are case insensitive, but
+				// default to upper case. In particular, the Spanner AST uses
+				// upper case, so spType is upper case. Many source DBs
+				// default to lower case. When printing source DB and
 				// Spanner types for comparison purposes, this can be distracting.
 				// Hence we switch to lower-case for Spanner types here.
-				sType = strings.ToLower(sType)
+				// TODO: add logic to choose case for Spanner types based
+				// on case of srcType.
+				spType = strings.ToLower(spType)
 				switch i {
 				case defaultValue:
-					l = append(l, fmt.Sprintf("%s e.g. column '%s'", issueDB[i].brief, pgCol))
+					l = append(l, fmt.Sprintf("%s e.g. column '%s'", issueDB[i].brief, srcCol))
 				case foreignKey:
-					l = append(l, fmt.Sprintf("Column '%s' uses foreign keys which Spanner does not support", pgCol))
+					l = append(l, fmt.Sprintf("Column '%s' uses foreign keys which Spanner does not support", srcCol))
 				case timestamp:
 					// Avoid the confusing "timestamp is mapped to timestamp" message.
-					l = append(l, fmt.Sprintf("Some columns have PostgreSQL type 'timestamp without timezone' which is mapped to Spanner type timestamp e.g. column '%s'. %s", pgCol, issueDB[i].brief))
+					l = append(l, fmt.Sprintf("Some columns have source DB type 'timestamp without timezone' which is mapped to Spanner type timestamp e.g. column '%s'. %s", srcCol, issueDB[i].brief))
 				case widened:
-					l = append(l, fmt.Sprintf("%s e.g. for column '%s', PostgreSQL type %s is mapped to Spanner type %s", issueDB[i].brief, pgCol, pgType, sType))
+					l = append(l, fmt.Sprintf("%s e.g. for column '%s', source DB type %s is mapped to Spanner type %s", issueDB[i].brief, srcCol, srcType, spType))
 				default:
-					l = append(l, fmt.Sprintf("Column '%s': type %s is mapped to %s. %s", pgCol, pgType, sType, issueDB[i].brief))
+					l = append(l, fmt.Sprintf("Column '%s': type %s is mapped to %s. %s", srcCol, srcType, spType, issueDB[i].brief))
 				}
 			}
 		}
@@ -199,18 +201,18 @@ func buildTableReportBody(conv *Conv, pgTable string, issues map[string][]schema
 	return body
 }
 
-func fillRowStats(conv *Conv, spTable string, badWrites map[string]int64, tr *tableReport) {
-	rows := conv.stats.rows[spTable]
-	goodConvRows := conv.stats.goodRows[spTable]
-	badConvRows := conv.stats.badRows[spTable]
-	badRowWrites := badWrites[spTable]
+func fillRowStats(conv *Conv, srcTable string, badWrites map[string]int64, tr *tableReport) {
+	rows := conv.stats.rows[srcTable]
+	goodConvRows := conv.stats.goodRows[srcTable]
+	badConvRows := conv.stats.badRows[srcTable]
+	badRowWrites := badWrites[srcTable]
 	// Note on rows:
 	// rows: all rows we encountered during processing.
 	// goodConvRows: rows we successfully converted.
 	// badConvRows: rows we failed to convert.
 	// badRowWrites: rows we converted, but could not write to Spanner.
 	if rows != goodConvRows+badConvRows || badRowWrites > goodConvRows {
-		conv.unexpected(fmt.Sprintf("Inconsistent row counts for table %s: %d %d %d %d\n", spTable, rows, goodConvRows, badConvRows, badRowWrites))
+		conv.unexpected(fmt.Sprintf("Inconsistent row counts for table %s: %d %d %d %d\n", srcTable, rows, goodConvRows, badConvRows, badRowWrites))
 	}
 	tr.rows = rows
 	tr.badRows = badConvRows + badRowWrites
@@ -248,9 +250,9 @@ const (
 )
 
 // analyzeCols returns information about the quality of schema mappings
-// for table 'pgTable'. It assumes 'pgTable' is in the conv.pgSchema map.
-func analyzeCols(conv *Conv, pgTable, spTable string) (map[string][]schemaIssue, int64, int64) {
-	pgSchema := conv.pgSchema[pgTable]
+// for table 'srcTable'. It assumes 'srcTable' is in the conv.srcSchema map.
+func analyzeCols(conv *Conv, srcTable, spTable string) (map[string][]schemaIssue, int64, int64) {
+	srcSchema := conv.srcSchema[srcTable]
 	m := make(map[string][]schemaIssue)
 	warnings := int64(0)
 	warningBatcher := make(map[schemaIssue]bool)
@@ -258,10 +260,10 @@ func analyzeCols(conv *Conv, pgTable, spTable string) (map[string][]schemaIssue,
 	// per column and/or multiple warnings per table.
 	// non-batched warnings: count at most one warning per column.
 	// batched warnings: count at most one warning per table.
-	for c, pc := range pgSchema.cols {
+	for c, l := range conv.issues[srcTable] {
 		colWarning := false
-		m[c] = pc.issues
-		for _, i := range pc.issues {
+		m[c] = l
+		for _, i := range l {
 			switch {
 			case issueDB[i].severity == warning && issueDB[i].batch:
 				warningBatcher[i] = true
@@ -274,14 +276,14 @@ func analyzeCols(conv *Conv, pgTable, spTable string) (map[string][]schemaIssue,
 		}
 	}
 	warnings += int64(len(warningBatcher))
-	return m, int64(len(pgSchema.cols)), warnings
+	return m, int64(len(srcSchema.ColDefs)), warnings
 }
 
-// rateSchema returns an string summarizing the quality of PostgreSQL
+// rateSchema returns an string summarizing the quality of source DB
 // to Spanner schema conversion. 'cols' and 'warnings' are respectively
 // the number of columns converted and the warnings encountered
 // (both weighted by number of data rows).
-// 'missingPKey' indicates whether the PostgreSQL schema had a primary key.
+// 'missingPKey' indicates whether the source DB schema had a primary key.
 // 'summary' indicates whether this is a per-table rating or an overall
 // summary rating.
 func rateSchema(cols, warnings int64, missingPKey, summary bool) string {
diff --git a/internal/postgres/tablemap.go b/internal/tablemap.go
similarity index 55%
rename from internal/postgres/tablemap.go
rename to internal/tablemap.go
index dd9d8bb..16feb86 100644
--- a/internal/postgres/tablemap.go
+++ b/internal/tablemap.go
@@ -12,32 +12,30 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"fmt"
 	"strconv"
-
-	"github.com/cloudspannerecosystem/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
+// GetSpannerTable maps a source DB table name into a legal Spanner table
+// name. Note that source DB 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 == "" {
+func GetSpannerTable(conv *Conv, srcTable string) (string, error) {
+	if srcTable == "" {
 		return "", fmt.Errorf("bad parameter: table string is empty")
 	}
-	if sp, found := conv.toSpanner[pgTable]; found {
+	if sp, found := conv.toSpanner[srcTable]; found {
 		return sp.name, nil
 	}
-	spTable, _ := internal.FixName(pgTable)
-	if _, found := conv.toPostgres[spTable]; found {
+	spTable, _ := FixName(srcTable)
+	if _, found := conv.toSource[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,
@@ -45,55 +43,55 @@ func GetSpannerTable(conv *Conv, pgTable string) (string, error) {
 		id := len(conv.toSpanner)
 		for {
 			t := spTable + "_" + strconv.Itoa(id)
-			if _, found := conv.toPostgres[t]; !found {
+			if _, found := conv.toSource[t]; !found {
 				spTable = t
 				break
 			}
 			id++
 		}
 	}
-	if spTable != pgTable {
-		internal.VerbosePrintf("Mapping PostgreSQL table %s to Spanner table %s\n", pgTable, spTable)
+	if spTable != srcTable {
+		VerbosePrintf("Mapping source DB table %s to Spanner table %s\n", srcTable, spTable)
 	}
-	conv.toSpanner[pgTable] = nameAndCols{name: spTable, cols: make(map[string]string)}
-	conv.toPostgres[spTable] = nameAndCols{name: pgTable, cols: make(map[string]string)}
+	conv.toSpanner[srcTable] = nameAndCols{name: spTable, cols: make(map[string]string)}
+	conv.toSource[spTable] = nameAndCols{name: srcTable, cols: make(map[string]string)}
 	return spTable, nil
 }
 
-// GetSpannerCol maps a PostgreSQL table/column into a legal Spanner column
+// GetSpannerCol maps a source DB 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
+// Note that source DB 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 == "" {
+func GetSpannerCol(conv *Conv, srcTable, srcCol string, mustExist bool) (string, error) {
+	if srcTable == "" {
 		return "", fmt.Errorf("bad parameter: table string is empty")
 	}
-	if pgCol == "" {
+	if srcCol == "" {
 		return "", fmt.Errorf("bad parameter: col string is empty")
 	}
-	sp, found := conv.toSpanner[pgTable]
+	sp, found := conv.toSpanner[srcTable]
 	if !found {
-		return "", fmt.Errorf("unknown table %s", pgTable)
+		return "", fmt.Errorf("unknown table %s", srcTable)
 	}
 	// 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)
+	src, found := conv.toSource[sp.name]
+	if !found || src.name != srcTable {
+		return "", fmt.Errorf("internal error: table mapping inconsistency for table %s (%s)", srcTable, src.name)
 	}
-	if spCol, found := sp.cols[pgCol]; found {
+	if spCol, found := sp.cols[srcCol]; found {
 		return spCol, nil
 	}
 	if mustExist {
-		return "", fmt.Errorf("table %s does not have a column %s", pgTable, pgCol)
+		return "", fmt.Errorf("table %s does not have a column %s", srcTable, srcCol)
 	}
-	spCol, _ := internal.FixName(pgCol)
-	if _, found := conv.toPostgres[sp.name].cols[spCol]; found {
+	spCol, _ := FixName(srcCol)
+	if _, found := conv.toSource[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,
@@ -101,17 +99,31 @@ func GetSpannerCol(conv *Conv, pgTable, pgCol string, mustExist bool) (string, e
 		id := len(sp.cols)
 		for {
 			c := spCol + "_" + strconv.Itoa(id)
-			if _, found := conv.toPostgres[sp.name].cols[c]; !found {
+			if _, found := conv.toSource[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)
+	if spCol != srcCol {
+		VerbosePrintf("Mapping source DB col %s (table %s) to Spanner col %s\n", srcCol, srcTable, spCol)
 	}
-	conv.toSpanner[pgTable].cols[pgCol] = spCol
-	conv.toPostgres[sp.name].cols[spCol] = pgCol
+	conv.toSpanner[srcTable].cols[srcCol] = spCol
+	conv.toSource[sp.name].cols[spCol] = srcCol
 	return spCol, nil
 }
+
+// GetSpannerCols maps a slice of source columns into their corresponding
+// Spanner columns using GetSpannerCol.
+func GetSpannerCols(conv *Conv, srcTable string, srcCols []string) ([]string, error) {
+	var spCols []string
+	for _, srcCol := range srcCols {
+		spCol, err := GetSpannerCol(conv, srcTable, srcCol, false)
+		if err != nil {
+			return nil, err
+		}
+		spCols = append(spCols, spCol)
+	}
+	return spCols, nil
+}
diff --git a/internal/toddl.go b/internal/toddl.go
new file mode 100644
index 0000000..7bfffd7
--- /dev/null
+++ b/internal/toddl.go
@@ -0,0 +1,204 @@
+// 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 internal
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+	"unicode"
+
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+// schemaToDDL performs schema conversion from the source DB schema to
+// Spanner. It uses the source schema in conv.srcSchema, and writes
+// the Spanner schema to conv.spSchema.
+func schemaToDDL(conv *Conv) error {
+	for _, srcTable := range conv.srcSchema {
+		spTableName, err := GetSpannerTable(conv, srcTable.Name)
+		if err != nil {
+			conv.unexpected(fmt.Sprintf("Couldn't map source table %s to Spanner: %s", srcTable.Name, err))
+			continue
+		}
+		var spColNames []string
+		spColDef := make(map[string]ddl.ColumnDef)
+		conv.issues[srcTable.Name] = make(map[string][]schemaIssue)
+		// Iterate over columns using ColNames order.
+		for _, srcColName := range srcTable.ColNames {
+			srcCol := srcTable.ColDefs[srcColName]
+			colName, err := GetSpannerCol(conv, srcTable.Name, srcCol.Name, false)
+			if err != nil {
+				conv.unexpected(fmt.Sprintf("Couldn't map source column %s of table %s to Spanner: %s", srcTable.Name, srcCol.Name, err))
+				continue
+			}
+			spColNames = append(spColNames, colName)
+			ty, issues := toSpannerType(conv, srcCol.Type.Name, srcCol.Type.Mods)
+			if len(srcCol.Type.ArrayBounds) > 1 {
+				ty = ddl.String{Len: ddl.MaxLength{}}
+				issues = append(issues, multiDimensionalArray)
+			}
+			// TODO: add issues for all elements of srcCol.Ignored.
+			if srcCol.Ignored.ForeignKey {
+				issues = append(issues, foreignKey)
+			}
+			if srcCol.Ignored.Default {
+				issues = append(issues, defaultValue)
+			}
+			if len(issues) > 0 {
+				conv.issues[srcTable.Name][srcCol.Name] = issues
+			}
+			spColDef[colName] = ddl.ColumnDef{
+				Name:    colName,
+				T:       ty,
+				IsArray: len(srcCol.Type.ArrayBounds) == 1,
+				NotNull: srcCol.NotNull,
+				Comment: "From: " + quoteIfNeeded(srcCol.Name) + " " + printSourceType(srcCol.Type),
+			}
+		}
+		comment := "Spanner schema for source table " + quoteIfNeeded(srcTable.Name)
+		conv.spSchema[spTableName] = ddl.CreateTable{
+			Name:     spTableName,
+			ColNames: spColNames,
+			ColDefs:  spColDef,
+			Pks:      cvtPrimaryKeys(conv, srcTable.Name, srcTable.PrimaryKeys),
+			Comment:  comment}
+	}
+	return nil
+}
+
+// toSpannerType maps a scalar source schema type (defined by id and
+// mods) into a Spanner type. This is the core source-to-Spanner type
+// mapping.  toSpannerType returns the Spanner type and a list of type
+// conversion issues encountered.
+func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []schemaIssue) {
+	maxExpectedMods := func(n int) {
+		if len(mods) > n {
+			conv.unexpected(fmt.Sprintf("Found %d mods while processing type id=%s", len(mods), id))
+		}
+	}
+	switch id {
+	case "bool":
+		maxExpectedMods(0)
+		return ddl.Bool{}, nil
+	case "bigserial":
+		maxExpectedMods(0)
+		return ddl.Int64{}, []schemaIssue{serial}
+	case "bpchar": // Note: Postgres internal name for char is bpchar (aka blank padded char).
+		maxExpectedMods(1)
+		if len(mods) > 0 {
+			return ddl.String{Len: ddl.Int64Length{Value: mods[0]}}, nil
+		}
+		// Note: bpchar without length specifier is equivalent to bpchar(1)
+		return ddl.String{Len: ddl.Int64Length{Value: 1}}, nil
+	case "bytea":
+		maxExpectedMods(0)
+		return ddl.Bytes{Len: ddl.MaxLength{}}, nil
+	case "date":
+		maxExpectedMods(0)
+		return ddl.Date{}, nil
+	case "float8":
+		maxExpectedMods(0)
+		return ddl.Float64{}, nil
+	case "float4":
+		maxExpectedMods(0)
+		return ddl.Float64{}, []schemaIssue{widened}
+	case "int8":
+		maxExpectedMods(0)
+		return ddl.Int64{}, nil
+	case "int4":
+		maxExpectedMods(0)
+		return ddl.Int64{}, []schemaIssue{widened}
+	case "int2":
+		maxExpectedMods(0)
+		return ddl.Int64{}, []schemaIssue{widened}
+	case "numeric": // Map all numeric types to float64.
+		maxExpectedMods(2)
+		if len(mods) > 0 && mods[0] <= 15 {
+			// float64 can represent this numeric type faithfully.
+			// Note: int64 has 53 bits for mantissa, which is ~15.96
+			// decimal digits.
+			return ddl.Float64{}, []schemaIssue{numericThatFits}
+		}
+		return ddl.Float64{}, []schemaIssue{numeric}
+	case "serial":
+		maxExpectedMods(0)
+		return ddl.Int64{}, []schemaIssue{serial}
+	case "text":
+		maxExpectedMods(0)
+		return ddl.String{Len: ddl.MaxLength{}}, nil
+	case "timestamptz":
+		maxExpectedMods(1)
+		return ddl.Timestamp{}, nil
+	case "timestamp":
+		maxExpectedMods(1)
+		// Map timestamp without timezone to Spanner timestamp.
+		return ddl.Timestamp{}, []schemaIssue{timestamp}
+	case "varchar":
+		maxExpectedMods(1)
+		if len(mods) > 0 {
+			return ddl.String{Len: ddl.Int64Length{Value: mods[0]}}, nil
+		}
+		return ddl.String{Len: ddl.MaxLength{}}, nil
+	}
+	return ddl.String{Len: ddl.MaxLength{}}, []schemaIssue{noGoodType}
+}
+
+func printSourceType(ty schema.Type) string {
+	s := ty.Name
+	if len(ty.Mods) > 0 {
+		var l []string
+		for _, x := range ty.Mods {
+			l = append(l, strconv.FormatInt(x, 10))
+		}
+		s = fmt.Sprintf("%s(%s)", s, strings.Join(l, ","))
+	}
+	if len(ty.ArrayBounds) > 0 {
+		l := []string{s}
+		for _, x := range ty.ArrayBounds {
+			if x == -1 {
+				l = append(l, "[]")
+			} else {
+				l = append(l, fmt.Sprintf("[%d]", x))
+			}
+		}
+		s = strings.Join(l, "")
+	}
+	return s
+}
+
+func quoteIfNeeded(s string) string {
+	for _, r := range s {
+		if unicode.IsLetter(r) || unicode.IsDigit(r) || unicode.IsPunct(r) {
+			continue
+		}
+		return strconv.Quote(s)
+	}
+	return s
+}
+
+func cvtPrimaryKeys(conv *Conv, srcTable string, srcKeys []schema.Key) []ddl.IndexKey {
+	var spKeys []ddl.IndexKey
+	for _, k := range srcKeys {
+		spCol, err := GetSpannerCol(conv, srcTable, k.Column, true)
+		if err != nil {
+			conv.unexpected(fmt.Sprintf("Can't map key for table %s", srcTable))
+			continue
+		}
+		spKeys = append(spKeys, ddl.IndexKey{Col: spCol, Desc: k.Desc})
+	}
+	return spKeys
+}
diff --git a/main.go b/main.go
index db2aa44..8989606 100644
--- a/main.go
+++ b/main.go
@@ -41,7 +41,6 @@ import (
 	instancepb "google.golang.org/genproto/googleapis/spanner/admin/instance/v1"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
-	"github.com/cloudspannerecosystem/harbourbridge/internal/postgres"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
@@ -140,7 +139,7 @@ func process(projectID, instanceID, dbName string, helper *ioStreams, ouputFileP
 		return fmt.Errorf("can't get seekable input file")
 	}
 	defer f.Close()
-	conv := postgres.MakeConv()
+	conv := internal.MakeConv()
 
 	err = firstPass(f, n, conv)
 	if err != nil {
@@ -171,7 +170,7 @@ func process(projectID, instanceID, dbName string, helper *ioStreams, ouputFileP
 	return nil
 }
 
-func report(bw *spanner.BatchWriter, bytesRead int64, now time.Time, db string, conv *postgres.Conv, reportFileName string, badDataFileName string) {
+func report(bw *spanner.BatchWriter, bytesRead int64, now time.Time, db string, conv *internal.Conv, reportFileName string, badDataFileName string) {
 	f, err := os.Create(reportFileName)
 	if err != nil {
 		fmt.Fprintf(ioHelper.out, "Can't write out report file %s: %v\n", reportFileName, err)
@@ -184,7 +183,7 @@ func report(bw *spanner.BatchWriter, bytesRead int64, now time.Time, db string,
 	banner := fmt.Sprintf("Generated at %s for db %s\n\n", now.Format("2006-01-02 15:04:05"), db)
 	w.WriteString(banner)
 	badWrites := bw.DroppedRowsByTable()
-	summary := postgres.GenerateReport(conv, w, badWrites)
+	summary := internal.GenerateReport(conv, w, badWrites)
 	w.Flush()
 	fmt.Fprintf(ioHelper.out, "Processed %d bytes of pg_dump data (%d statements, %d rows of data, %d errors, %d unexpected conditions).\n",
 		bytesRead, conv.Statements(), conv.Rows(), conv.StatementErrors(), conv.Unexpecteds())
@@ -197,12 +196,12 @@ func report(bw *spanner.BatchWriter, bytesRead int64, now time.Time, db string,
 	writeBadData(bw, conv, banner, badDataFileName)
 }
 
-func firstPass(f *os.File, fileSize int64, conv *postgres.Conv) error {
+func firstPass(f *os.File, fileSize int64, conv *internal.Conv) error {
 	p := internal.NewProgress(fileSize, "Generating schema", internal.Verbose())
 	r := internal.NewReader(bufio.NewReader(f), p)
 	conv.SetSchemaMode() // Build schema and ignore data in pg_dump.
 	conv.SetDataSink(nil)
-	err := postgres.ProcessPgDump(conv, r)
+	err := internal.ProcessPgDump(conv, r)
 	if err != nil {
 		return err
 	}
@@ -210,7 +209,7 @@ func firstPass(f *os.File, fileSize int64, conv *postgres.Conv) error {
 	return nil
 }
 
-func secondPass(f *os.File, client *sp.Client, conv *postgres.Conv, totalRows int64) *spanner.BatchWriter {
+func secondPass(f *os.File, client *sp.Client, conv *internal.Conv, totalRows int64) *spanner.BatchWriter {
 	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
 	r := internal.NewReader(bufio.NewReader(f), nil)
 	rows := int64(0)
@@ -235,7 +234,7 @@ func secondPass(f *os.File, client *sp.Client, conv *postgres.Conv, totalRows in
 		func(table string, cols []string, vals []interface{}) {
 			writer.AddRow(table, cols, vals)
 		})
-	postgres.ProcessPgDump(conv, r)
+	internal.ProcessPgDump(conv, r)
 	writer.Flush()
 	p.Done()
 	return writer
@@ -276,7 +275,7 @@ func getSeekable(f *os.File) (*os.File, int64, error) {
 // It automatically determines an appropriate project, selects a
 // Spanner instance to use, generates a new Spanner DB name,
 // and call into the Spanner admin interface to create the new DB.
-func createDatabase(project, instance, dbName string, conv *postgres.Conv) (string, error) {
+func createDatabase(project, instance, dbName string, conv *internal.Conv) (string, error) {
 	fmt.Fprintf(ioHelper.out, "Creating new database %s in instance %s with default permissions ... ", dbName, instance)
 	ctx := context.Background()
 	adminClient, err := database.NewDatabaseAdminClient(ctx)
@@ -369,7 +368,7 @@ func getInstances(project string) ([]string, error) {
 	return l, nil
 }
 
-func writeSchemaFile(conv *postgres.Conv, now time.Time, name string) {
+func writeSchemaFile(conv *internal.Conv, now time.Time, name string) {
 	f, err := os.Create(name)
 	if err != nil {
 		fmt.Fprintf(ioHelper.out, "Can't create schema file %s: %v\n", name, err)
@@ -399,7 +398,7 @@ func writeSchemaFile(conv *postgres.Conv, now time.Time, name string) {
 
 // writeBadData prints summary stats about bad rows and writes detailed info
 // to file 'name'.
-func writeBadData(bw *spanner.BatchWriter, conv *postgres.Conv, banner, name string) {
+func writeBadData(bw *spanner.BatchWriter, conv *internal.Conv, banner, name string) {
 	badConversions := conv.BadRows()
 	badWrites := sum(bw.DroppedRowsByTable())
 	if badConversions == 0 && badWrites == 0 {
diff --git a/schema/schema.go b/schema/schema.go
new file mode 100644
index 0000000..e7d9219
--- /dev/null
+++ b/schema/schema.go
@@ -0,0 +1,75 @@
+// 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 schema provides a generic representation of database
+// schemas. Note that our goal is not to faithfully represent all
+// aspects of the schema, but just the relevant components for
+// conversion to Spanner and reporting on the quality of the
+// conversion (this motivates us to keep partial information about
+// some features we will report on but not use in the conversion
+// e.g. default values, check constraints).
+//
+// The current version supports PostgreSQL. Expect it to grow as we
+// support other databases. We might eventually support the Spanner
+// schema, and potentially get rid of the ddl package.
+package schema
+
+// Table represents a database table.
+type Table struct {
+	Name        string
+	ColNames    []string          // List of column names (for predictable iteration order e.g. printing).
+	ColDefs     map[string]Column // Details of columns.
+	PrimaryKeys []Key
+	Indexes     []Index
+}
+
+// Column represents a database column.
+// TODO: add support for foreign keys.
+type Column struct {
+	Name    string
+	Type    Type
+	NotNull bool
+	Unique  bool
+	Ignored Ignored
+}
+
+// Key respresents a primary key or index key.
+type Key struct {
+	Column string
+	Desc   bool // By default, order is ASC. Set to true to specifiy DESC.
+}
+
+// Index represents a database index.
+type Index struct {
+	Name string
+	Keys []Key
+}
+
+// Type represents the type of a column.
+type Type struct {
+	Name        string
+	Mods        []int64 // List of modifiers (aka type parameters e.g. varchar(8) or numeric(6, 4).
+	ArrayBounds []int64 // Empty for scalar types.
+}
+
+// Ignored represents column properties/constraints that are not
+// represented. We drop the details, but retain presence/absence for
+// reporting purposes.
+type Ignored struct {
+	Check      bool
+	Identity   bool
+	Default    bool
+	Exclusion  bool
+	ForeignKey bool
+}
diff --git a/spanner/ddl/ast.go b/spanner/ddl/ast.go
index 2f3c1e2..83f150f 100644
--- a/spanner/ddl/ast.go
+++ b/spanner/ddl/ast.go
@@ -185,11 +185,11 @@ func (pk IndexKey) PrintIndexKey(c Config) string {
 // CreateTable encodes the following DDL definition:
 //     create_table: CREATE TABLE table_name ([column_def, ...] ) primary_key [, cluster]
 type CreateTable struct {
-	Name    string
-	Cols    []string             // Provides names and order of columns
-	Cds     map[string]ColumnDef // Provides definition of columns (a map for simpler/faster lookup during type processing)
-	Pks     []IndexKey
-	Comment string
+	Name     string
+	ColNames []string             // Provides names and order of columns
+	ColDefs  map[string]ColumnDef // Provides definition of columns (a map for simpler/faster lookup during type processing)
+	Pks      []IndexKey
+	Comment  string
 }
 
 // PrintCreateTable unparses a CREATE TABLE statement.
@@ -197,10 +197,10 @@ func (ct CreateTable) PrintCreateTable(config Config) string {
 	var col []string
 	var colComment []string
 	var keys []string
-	for i, cn := range ct.Cols {
-		s, c := ct.Cds[cn].PrintColumnDef(config)
+	for i, cn := range ct.ColNames {
+		s, c := ct.ColDefs[cn].PrintColumnDef(config)
 		s = "\n    " + s
-		if i < len(ct.Cols)-1 {
+		if i < len(ct.ColNames)-1 {
 			s += ","
 		} else {
 			s += " "


Test patch:
diff --git a/internal/postgres/convert_test.go b/internal/convert_test.go
similarity index 99%
rename from internal/postgres/convert_test.go
rename to internal/convert_test.go
index adf577e..17fb46f 100644
--- a/internal/postgres/convert_test.go
+++ b/internal/convert_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"strings"
diff --git a/internal/postgres/data_test.go b/internal/data_test.go
similarity index 56%
rename from internal/postgres/data_test.go
rename to internal/data_test.go
index c39ff22..30a9340 100644
--- a/internal/postgres/data_test.go
+++ b/internal/data_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"fmt"
@@ -23,51 +23,88 @@ import (
 	"cloud.google.com/go/civil"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
+// Basic smoke test of ProcessDataRow. The core part of this code path
+// (ConvertData) is tested in TestConvertData.
+func TestProcessDataRow(t *testing.T) {
+	tableName := "testtable"
+	cols := []string{"a", "b", "c"}
+	conv := buildConv(
+		ddl.CreateTable{
+			Name:     tableName,
+			ColNames: cols,
+			ColDefs: map[string]ddl.ColumnDef{
+				"a": ddl.ColumnDef{Name: "a", T: ddl.Float64{}},
+				"b": ddl.ColumnDef{Name: "b", T: ddl.Int64{}},
+				"c": ddl.ColumnDef{Name: "c", T: ddl.String{Len: ddl.MaxLength{}}},
+			}},
+		schema.Table{
+			Name:     tableName,
+			ColNames: cols,
+			ColDefs: map[string]schema.Column{
+				"a": schema.Column{Name: "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})
+		})
+	ProcessDataRow(conv, tableName, cols, []string{"4.2", "6", "prisoner zero"})
+	assert.Equal(t, []spannerData{spannerData{table: tableName, cols: cols, vals: []interface{}{float64(4.2), int64(6), "prisoner zero"}}}, rows)
+}
+
 func TestConvertData(t *testing.T) {
 	singleColTests := []struct {
 		name    string
-		t       ddl.ScalarType
+		ty      ddl.ScalarType
 		isArray bool
-		pgt     string      // Postgres type (used by e.g. timestamp conversions).
+		srcTy   string      // Source DB 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}},
+		{"bytes", ddl.Bytes{Len: 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"},
+		{"string", ddl.String{Len: 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}}},
+		{"bytes array", ddl.Bytes{Len: 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"}},
+		{"string array", ddl.String{Len: 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")}},
 	}
+	tableName := "testtable"
 	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 := buildConv(
+			ddl.CreateTable{
+				Name:     tableName,
+				ColNames: []string{col},
+				ColDefs:  map[string]ddl.ColumnDef{col: ddl.ColumnDef{Name: col, T: tc.ty, IsArray: tc.isArray, NotNull: false}},
+				Pks:      []ddl.IndexKey{}},
+			schema.Table{Name: tableName, ColNames: []string{col}, ColDefs: map[string]schema.Column{col: schema.Column{Type: schema.Type{Name: tc.srcTy}}}})
 		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)
+		at, ac, av, err := ConvertData(conv, tableName, []string{col}, []string{tc.in})
+		checkResults(t, at, ac, av, err, tableName, []string{col}, []interface{}{tc.e}, tc.name)
 	}
 
 	timestampTests := []struct {
-		name string
-		pgt  string
-		in   string
-		e    interface{}
+		name  string
+		srcTy 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")},
@@ -75,15 +112,21 @@ func TestConvertData(t *testing.T) {
 		{"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}}})
+		conv := buildConv(
+			ddl.CreateTable{
+				Name:     tableName,
+				ColNames: []string{col},
+				ColDefs:  map[string]ddl.ColumnDef{col: ddl.ColumnDef{Name: col, T: ddl.Timestamp{}}}},
+			schema.Table{
+				Name:     tableName,
+				ColNames: []string{col},
+				ColDefs:  map[string]schema.Column{col: schema.Column{Type: schema.Type{Name: tc.srcTy}}}})
 		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})
+		atable, ac, av, err := ConvertData(conv, tableName, []string{col}, []string{tc.in})
 		assert.Nil(t, err, tc.name)
+		assert.Equal(t, atable, tableName, tc.name+": table mismatch")
 		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
@@ -131,27 +174,26 @@ func TestConvertData(t *testing.T) {
 			evals: []interface{}{int64(6)},
 		},
 	}
-	spSchema := ddl.CreateTable{
-		"testtable",
-		[]string{"a", "b", "c"},
-		map[string]ddl.ColumnDef{
+	spTable := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: 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"},
+		}}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: []string{"a", "b", "c"},
+		ColDefs: map[string]schema.Column{
+			"a": schema.Column{Type: schema.Type{Name: "int8"}},
+			"b": schema.Column{Type: schema.Type{Name: "float8"}},
+			"c": schema.Column{Type: schema.Type{Name: "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)
+		conv := buildConv(spTable, srcTable)
+		atable, acols, avals, err := ConvertData(conv, spTable.Name, tc.cols, tc.vals)
+		checkResults(t, atable, acols, avals, err, tableName, tc.ecols, tc.evals, tc.name)
 	}
 
 	errorTests := []struct {
@@ -176,11 +218,9 @@ func TestConvertData(t *testing.T) {
 		},
 	}
 	for _, tc := range errorTests {
-		conv := buildConv(spSchema.Name, spSchema, pgSchema)
-		acols, avals, err := ConvertData(conv, spSchema.Name, spSchema.Name, tc.cols, tc.vals)
+		conv := buildConv(spTable, srcTable)
+		_, _, _, err := ConvertData(conv, spTable.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 {
@@ -205,23 +245,30 @@ func TestConvertData(t *testing.T) {
 			evals: []interface{}{int64(7), int64(bits.Reverse64(1))},
 		},
 	}
-	conv := buildConv(spSchema.Name, spSchema, pgSchema)
-	conv.syntheticPKeys[spSchema.Name] = syntheticPKey{col: "synth_id", sequence: 0}
+	conv := buildConv(spTable, srcTable)
+	conv.syntheticPKeys[spTable.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)
+		atable, acols, avals, err := ConvertData(conv, spTable.Name, tc.cols, tc.vals)
+		checkResults(t, atable, acols, avals, err, tableName, tc.ecols, tc.evals, tc.name)
 	}
 }
 
-func buildConv(table string, spSchema ddl.CreateTable, pgSchema pgTableDef) *Conv {
+func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *Conv {
 	conv := MakeConv()
-	conv.spSchema[table] = spSchema
-	conv.pgSchema[table] = pgSchema
+	conv.spSchema[spTable.Name] = spTable
+	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
+	}
 	return conv
 }
 
-func checkResults(t *testing.T, acols []string, avals []interface{}, err error, ecols []string, evals []interface{}, name string) {
+func checkResults(t *testing.T, atable string, acols []string, avals []interface{}, err error, etable string, ecols []string, evals []interface{}, name string) {
 	assert.Nil(t, err, name)
+	assert.Equal(t, atable, etable, name+": table mismatch")
 	assert.Equal(t, ecols, acols, name+": column mismatch")
 	assert.Equal(t, evals, avals, name+": value mismatch")
 }
diff --git a/internal/postgres/process_test.go b/internal/pgdump_test.go
similarity index 93%
rename from internal/postgres/process_test.go
rename to internal/pgdump_test.go
index 905c318..93e0750 100644
--- a/internal/postgres/process_test.go
+++ b/internal/pgdump_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"bufio"
@@ -25,7 +25,6 @@ import (
 	pg_query "github.com/lfittl/pg_query_go"
 	"github.com/stretchr/testify/assert"
 
-	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
@@ -70,7 +69,7 @@ func TestProcessPgDump(t *testing.T) {
 	for _, tc := range scalarTests {
 		conv, _ := runProcessPgDump(fmt.Sprintf("CREATE TABLE t (a %s);", tc.ty))
 		noIssues(conv, t, "Scalar type: "+tc.ty)
-		assert.Equal(t, conv.spSchema["t"].Cds["a"].T, tc.expected, "Scalar type: "+tc.ty)
+		assert.Equal(t, conv.spSchema["t"].ColDefs["a"].T, tc.expected, "Scalar type: "+tc.ty)
 	}
 	// Next test array types and not null.
 	singleColTests := []struct {
@@ -87,7 +86,7 @@ func TestProcessPgDump(t *testing.T) {
 	for _, tc := range singleColTests {
 		conv, _ := runProcessPgDump(fmt.Sprintf("CREATE TABLE t (a %s);", tc.ty))
 		noIssues(conv, t, "Not null: "+tc.ty)
-		cd := conv.spSchema["t"].Cds["a"]
+		cd := conv.spSchema["t"].ColDefs["a"]
 		cd.Comment = ""
 		assert.Equal(t, tc.expected, cd, "Not null: "+tc.ty)
 	}
@@ -105,9 +104,9 @@ func TestProcessPgDump(t *testing.T) {
 				"ALTER TABLE ONLY cart ADD CONSTRAINT cart_pkey PRIMARY KEY (productid, userid);\n",
 			expectedSchema: map[string]ddl.CreateTable{
 				"cart": ddl.CreateTable{
-					Name: "cart",
-					Cols: []string{"productid", "userid", "quantity"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "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{}},
@@ -119,9 +118,9 @@ func TestProcessPgDump(t *testing.T) {
 			input: "CREATE TABLE cart (productid text, userid text NOT NULL, quantity bigint);\n",
 			expectedSchema: map[string]ddl.CreateTable{
 				"cart": ddl.CreateTable{
-					Name: "cart",
-					Cols: []string{"productid", "userid", "quantity", "synth_id"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "cart",
+					ColNames: []string{"productid", "userid", "quantity", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
 						"productid": ddl.ColumnDef{Name: "productid", T: ddl.String{Len: ddl.MaxLength{}}},
 						"userid":    ddl.ColumnDef{Name: "userid", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true},
 						"quantity":  ddl.ColumnDef{Name: "quantity", T: ddl.Int64{}},
@@ -134,9 +133,9 @@ func TestProcessPgDump(t *testing.T) {
 			input: "CREATE TABLE test (a text PRIMARY KEY, b text);\n",
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
-					Name: "test",
-					Cols: []string{"a", "b"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "test",
+					ColNames: []string{"a", "b"},
+					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.String{Len: ddl.MaxLength{}}},
 					},
@@ -147,9 +146,9 @@ func TestProcessPgDump(t *testing.T) {
 			input: "CREATE TABLE test (a text, b text, n bigint, PRIMARY KEY (a, b) );\n",
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
-					Name: "test",
-					Cols: []string{"a", "b", "n"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "test",
+					ColNames: []string{"a", "b", "n"},
+					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.String{Len: ddl.MaxLength{}}, NotNull: true},
 						"n": ddl.ColumnDef{Name: "n", T: ddl.Int64{}},
@@ -161,9 +160,9 @@ func TestProcessPgDump(t *testing.T) {
 			input: "CREATE TABLE myschema.test (a text PRIMARY KEY, b text);\n",
 			expectedSchema: map[string]ddl.CreateTable{
 				"myschema_test": ddl.CreateTable{
-					Name: "myschema_test",
-					Cols: []string{"a", "b"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "myschema_test",
+					ColNames: []string{"a", "b"},
+					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.String{Len: ddl.MaxLength{}}},
 					},
@@ -175,9 +174,9 @@ func TestProcessPgDump(t *testing.T) {
 				"ALTER TABLE test ALTER COLUMN b SET NOT NULL;\n",
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
-					Name: "test",
-					Cols: []string{"a", "b"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "test",
+					ColNames: []string{"a", "b"},
+					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.String{Len: ddl.MaxLength{}}, NotNull: true},
 					},
@@ -190,16 +189,16 @@ func TestProcessPgDump(t *testing.T) {
 				"ALTER TABLE ONLY t2 ADD CONSTRAINT t2_pkey PRIMARY KEY (c);",
 			expectedSchema: map[string]ddl.CreateTable{
 				"t1": ddl.CreateTable{
-					Name: "t1",
-					Cols: []string{"a", "b"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "t1",
+					ColNames: []string{"a", "b"},
+					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.String{Len: ddl.MaxLength{}}}},
 					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}}},
 				"t2": ddl.CreateTable{
-					Name: "t2",
-					Cols: []string{"c"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "t2",
+					ColNames: []string{"c"},
+					ColDefs: map[string]ddl.ColumnDef{
 						"c": ddl.ColumnDef{Name: "c", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true}},
 					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "c"}}}},
 		},
@@ -309,9 +308,9 @@ func TestProcessPgDump(t *testing.T) {
 				spannerData{table: "te_s_t", cols: []string{"a__", "b_b", "n_n"}, vals: []interface{}{"a", "b", int64(2)}}},
 			expectedSchema: map[string]ddl.CreateTable{
 				"te_s_t": ddl.CreateTable{
-					Name: "te_s_t",
-					Cols: []string{"a__", "b_b", "n_n"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "te_s_t",
+					ColNames: []string{"a__", "b_b", "n_n"},
+					ColDefs: map[string]ddl.ColumnDef{
 						"a__": ddl.ColumnDef{Name: "a__", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true},
 						"b_b": ddl.ColumnDef{Name: "b_b", T: ddl.String{Len: ddl.MaxLength{}}},
 						"n_n": ddl.ColumnDef{Name: "n_n", T: ddl.Int64{}},
@@ -327,9 +326,9 @@ func TestProcessPgDump(t *testing.T) {
 				spannerData{table: "te_s_t", cols: []string{"a__", "b_b", "n_n"}, vals: []interface{}{"a", "b", int64(2)}}},
 			expectedSchema: map[string]ddl.CreateTable{
 				"te_s_t": ddl.CreateTable{
-					Name: "te_s_t",
-					Cols: []string{"a__", "b_b", "n_n"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "te_s_t",
+					ColNames: []string{"a__", "b_b", "n_n"},
+					ColDefs: map[string]ddl.ColumnDef{
 						"a__": ddl.ColumnDef{Name: "a__", T: ddl.String{Len: ddl.MaxLength{}}, NotNull: true},
 						"b_b": ddl.ColumnDef{Name: "b_b", T: ddl.String{Len: ddl.MaxLength{}}},
 						"n_n": ddl.ColumnDef{Name: "n_n", T: ddl.Int64{}},
@@ -495,9 +494,9 @@ func TestProcessPgDump_AddPrimaryKeys(t *testing.T) {
 			input: "CREATE TABLE cart (productid text, userid text, quantity bigint);",
 			expectedSchema: map[string]ddl.CreateTable{
 				"cart": ddl.CreateTable{
-					Name: "cart",
-					Cols: []string{"productid", "userid", "quantity", "synth_id"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "cart",
+					ColNames: []string{"productid", "userid", "quantity", "synth_id"},
+					ColDefs: map[string]ddl.ColumnDef{
 						"productid": ddl.ColumnDef{Name: "productid", T: ddl.String{Len: ddl.MaxLength{}}},
 						"userid":    ddl.ColumnDef{Name: "userid", T: ddl.String{Len: ddl.MaxLength{}}},
 						"quantity":  ddl.ColumnDef{Name: "quantity", T: ddl.Int64{}},
@@ -510,9 +509,9 @@ func TestProcessPgDump_AddPrimaryKeys(t *testing.T) {
 			input: "CREATE TABLE test (synth_id text, synth_id0 text, synth_id1 bigint);",
 			expectedSchema: map[string]ddl.CreateTable{
 				"test": ddl.CreateTable{
-					Name: "test",
-					Cols: []string{"synth_id", "synth_id0", "synth_id1", "synth_id2"},
-					Cds: map[string]ddl.ColumnDef{
+					Name:     "test",
+					ColNames: []string{"synth_id", "synth_id0", "synth_id1", "synth_id2"},
+					ColDefs: map[string]ddl.ColumnDef{
 						"synth_id":  ddl.ColumnDef{Name: "synth_id", T: ddl.String{Len: ddl.MaxLength{}}},
 						"synth_id0": ddl.ColumnDef{Name: "synth_id0", T: ddl.String{Len: ddl.MaxLength{}}},
 						"synth_id1": ddl.ColumnDef{Name: "synth_id1", T: ddl.Int64{}},
@@ -535,7 +534,7 @@ func TestProcessPgDump_WithUnparsableContent(t *testing.T) {
 	conv := MakeConv()
 	conv.SetLocation(time.UTC)
 	conv.SetSchemaMode()
-	err := ProcessPgDump(conv, internal.NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	err := ProcessPgDump(conv, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
 	if err == nil {
 		t.Fatalf("Expect an error, but got nil")
 	}
@@ -548,14 +547,14 @@ func runProcessPgDump(s string) (*Conv, []spannerData) {
 	conv := MakeConv()
 	conv.SetLocation(time.UTC)
 	conv.SetSchemaMode()
-	ProcessPgDump(conv, internal.NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	ProcessPgDump(conv, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
 	conv.SetDataMode()
 	var rows []spannerData
 	conv.SetDataSink(
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	ProcessPgDump(conv, internal.NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	ProcessPgDump(conv, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
 	return conv, rows
 }
 
@@ -579,9 +578,9 @@ func noIssues(conv *Conv, t *testing.T, name string) {
 // are often changed and are not a core part of conversion functionality.
 func stripSchemaComments(spSchema map[string]ddl.CreateTable) map[string]ddl.CreateTable {
 	for t, ct := range spSchema {
-		for c, cd := range ct.Cds {
+		for c, cd := range ct.ColDefs {
 			cd.Comment = ""
-			ct.Cds[c] = cd
+			ct.ColDefs[c] = cd
 		}
 		ct.Comment = ""
 		spSchema[t] = ct
diff --git a/internal/postgres/statement_test.go b/internal/postgres/statement_test.go
deleted file mode 100644
index 627ddd5..0000000
--- a/internal/postgres/statement_test.go
+++ /dev/null
@@ -1,41 +0,0 @@
-// 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"
-
-	pg_query "github.com/lfittl/pg_query_go"
-	"github.com/stretchr/testify/assert"
-
-	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
-)
-
-// This is just a very basic smoke-test for processStatements.
-// The real testing of processStatements happens in process_test.go
-// via the public API ProcessPgDump (see TestProcessPgDump).
-func TestProcessStatements(t *testing.T) {
-	conv := MakeConv()
-	conv.SetSchemaMode()
-	s := "CREATE TABLE cart (productid text, userid text, quantity bigint);\n" +
-		"ALTER TABLE ONLY cart ADD CONSTRAINT cart_pkey PRIMARY KEY (productid, userid);\n"
-	tree, err := pg_query.Parse(s)
-	assert.Nil(t, err, "Failed to parse")
-	ci := processStatements(conv, tree.Statements)
-	assert.Nil(t, ci, "Unexpected COPY-FROM or INSERT")
-	assert.Equal(t, []string{"productid", "userid", "quantity"}, conv.spSchema["cart"].Cols)
-	assert.Equal(t, 3, len(conv.spSchema["cart"].Cds))
-	assert.Equal(t, []ddl.IndexKey{ddl.IndexKey{Col: "productid"}, ddl.IndexKey{Col: "userid"}}, conv.spSchema["cart"].Pks)
-}
diff --git a/internal/postgres/report_test.go b/internal/report_test.go
similarity index 95%
rename from internal/postgres/report_test.go
rename to internal/report_test.go
index d4994e0..9966fb9 100644
--- a/internal/postgres/report_test.go
+++ b/internal/report_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"bufio"
@@ -22,8 +22,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-
-	"github.com/cloudspannerecosystem/harbourbridge/internal"
 )
 
 func TestReport(t *testing.T) {
@@ -48,7 +46,7 @@ func TestReport(t *testing.T) {
             c text);`
 	conv := MakeConv()
 	conv.SetSchemaMode()
-	ProcessPgDump(conv, internal.NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	ProcessPgDump(conv, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
 	conv.stats.rows = map[string]int64{"bad_schema": 1000, "no_pk": 5000}
 	conv.stats.goodRows = map[string]int64{"bad_schema": 990, "no_pk": 3000}
 	conv.stats.badRows = map[string]int64{"bad_schema": 10, "no_pk": 2000}
@@ -109,8 +107,8 @@ Warnings
    type.
 
 Note
-1) Some columns will consume more storage in Spanner e.g. for column 'b',
-   PostgreSQL type int4 is mapped to Spanner type int64.
+1) Some columns will consume more storage in Spanner e.g. for column 'b', source
+   DB type int4 is mapped to Spanner type int64.
 
 ----------------------------
 Table default_value
@@ -148,8 +146,8 @@ Warning
    Spanner requires a primary key for every table.
 
 Note
-1) Some columns will consume more storage in Spanner e.g. for column 'b',
-   PostgreSQL type int4 is mapped to Spanner type int64.
+1) Some columns will consume more storage in Spanner e.g. for column 'b', source
+   DB type int4 is mapped to Spanner type int64.
 
 ----------------------------
 Unexpected Conditions
diff --git a/internal/postgres/tablemap_test.go b/internal/tablemap_test.go
similarity index 79%
rename from internal/postgres/tablemap_test.go
rename to internal/tablemap_test.go
index e057bb5..3965e5c 100644
--- a/internal/postgres/tablemap_test.go
+++ b/internal/tablemap_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package postgres
+package internal
 
 import (
 	"testing"
@@ -23,10 +23,10 @@ import (
 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.
+		name     string // Name of test.
+		srcTable string // Source DB table name to test.
+		error    bool   // Whether an error is expected.
+		spTable  string // Expected Spanner table name.
 	}{
 		{"Empty", "", true, ""},
 		{"Good name", "table", false, "table"},
@@ -41,14 +41,14 @@ func TestGetSpannerTable(t *testing.T) {
 		{"Illegal start character with collision (2)", "\ntable", false, "Atable_9"},
 	}
 	for _, tc := range basicTests {
-		spTable, err := GetSpannerTable(conv, tc.pgTable)
+		spTable, err := GetSpannerTable(conv, tc.srcTable)
 		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)
+		s2, err := GetSpannerTable(conv, tc.srcTable)
 		assert.Nil(t, err, tc.name)
 		assert.Equal(t, spTable, s2, tc.name)
 	}
@@ -57,11 +57,11 @@ func TestGetSpannerTable(t *testing.T) {
 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.
+		name     string // Name of test.
+		srcTable string // Source DB table name to test.
+		srcCol   string // Source DB 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, ""},
@@ -78,15 +78,15 @@ func TestGetSpannerCol(t *testing.T) {
 		{"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)
+		_, err1 := GetSpannerTable(conv, tc.srcTable) // Ensure table is known.
+		spCol, err2 := GetSpannerCol(conv, tc.srcTable, tc.srcCol, 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)
+		spCol2, err := GetSpannerCol(conv, tc.srcTable, tc.srcCol, false)
 		assert.Nil(t, err, tc.name)
 		assert.Equal(t, spCol, spCol2, tc.name)
 	}
diff --git a/internal/toddl_test.go b/internal/toddl_test.go
new file mode 100644
index 0000000..4ff18ee
--- /dev/null
+++ b/internal/toddl_test.go
@@ -0,0 +1,76 @@
+// 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 internal
+
+import (
+	"testing"
+
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
+)
+
+// This is just a very basic smoke-test for toSpannerType.
+// The real testing of toSpannerType happens in process_test.go
+// via the public API ProcessPgDump (see TestProcessPgDump).
+func TestToSpannerType(t *testing.T) {
+	conv := MakeConv()
+	conv.SetSchemaMode()
+	name := "test"
+	srcSchema := schema.Table{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f"},
+		ColDefs: map[string]schema.Column{
+			"a": schema.Column{Name: "a", Type: schema.Type{Name: "int8"}},
+			"b": schema.Column{Name: "b", Type: schema.Type{Name: "float4"}},
+			"c": schema.Column{Name: "c", Type: schema.Type{Name: "bool"}},
+			"d": schema.Column{Name: "d", Type: schema.Type{Name: "varchar", Mods: []int64{6}}},
+			"e": schema.Column{Name: "e", Type: schema.Type{Name: "numeric"}},
+			"f": schema.Column{Name: "f", Type: schema.Type{Name: "timestamptz"}},
+		},
+		PrimaryKeys: []schema.Key{schema.Key{Column: "a"}}}
+	conv.srcSchema[name] = srcSchema
+	assert.Nil(t, schemaToDDL(conv))
+	actual := conv.spSchema[name]
+	dropComments(&actual) // Don't test comment.
+	expected := ddl.CreateTable{
+		Name:     name,
+		ColNames: []string{"a", "b", "c", "d", "e", "f"},
+		ColDefs: 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{}},
+			"d": ddl.ColumnDef{Name: "d", T: ddl.String{Len: ddl.Int64Length{Value: 6}}},
+			"e": ddl.ColumnDef{Name: "e", T: ddl.Float64{}},
+			"f": ddl.ColumnDef{Name: "f", T: ddl.Timestamp{}},
+		},
+		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
+	}
+	assert.Equal(t, expected, actual)
+	expectedIssues := map[string][]schemaIssue{
+		"b": []schemaIssue{widened},
+		"e": []schemaIssue{numeric},
+	}
+	assert.Equal(t, expectedIssues, conv.issues[name])
+}
+
+func dropComments(t *ddl.CreateTable) {
+	t.Comment = ""
+	for _, c := range t.ColNames {
+		cd := t.ColDefs[c]
+		cd.Comment = ""
+		t.ColDefs[c] = cd
+	}
+}