Repository: cloudspannerecosystem/harbourbridge
Pull request #61 (2020-06-10 07:09:46)

Added postgres package
Moved postgres specific files to postgres package. Changes in files of `postgres` package to access functions() and variables from `internal` package.

googlebot: We found a Contributor License Agreement for you (the sender of this pull request), but were unable to find agreements for all the commit author(s) or Co-authors.  If you authored these, maybe you used a different email address in the git commits than was used to sign the CLA ([login here](https://cla.developers.google.com/) to double check)?  If these were authored by someone else, then they will need to sign a CLA as well, and confirm that they're okay with these being contributed to Google.
In order to pass this check, please resolve this problem and then comment `@googlebot I fixed it.`. If the bot doesn't comment, it means it doesn't think anything has changed.

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fcloudspannerecosystem%2Fharbourbridge%2Fpull%2F61) for more info**.

<!-- need_author_cla -->
Dharmil-searce: @googlebot I fixed it.
googlebot: CLAs look good, thanks!

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fcloudspannerecosystem%2Fharbourbridge%2Fpull%2F61) for more info**.

<!-- ok -->

Patch:
diff --git a/internal/report.go b/internal/report.go
index 77a6105..7f1a0a7 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -168,7 +168,7 @@ func buildTableReportBody(conv *Conv, srcTable string, issues map[string][]Schem
 				if err != nil {
 					conv.Unexpected(err.Error())
 				}
-				srcType := printSourceType(srcSchema.ColDefs[srcCol].Type)
+				srcType := srcSchema.ColDefs[srcCol].Type.Print()
 				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
diff --git a/main.go b/main.go
index 43c4c38..9d41af4 100644
--- a/main.go
+++ b/main.go
@@ -46,6 +46,7 @@ import (
 	"google.golang.org/grpc"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/postgres"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
@@ -243,7 +244,7 @@ func schemaFromSQL(driver string) (*internal.Conv, error) {
 		return nil, err
 	}
 	conv := internal.MakeConv()
-	err = internal.ProcessInfoSchema(conv, sourceDB)
+	err = postgres.ProcessInfoSchema(conv, sourceDB)
 	if err != nil {
 		return nil, err
 	}
@@ -265,7 +266,7 @@ func dataFromSQL(config spanner.BatchWriterConfig, client *sp.Client, conv *inte
 		return nil, err
 	}
 
-	internal.SetRowStats(conv, sourceDB)
+	postgres.SetRowStats(conv, sourceDB)
 	totalRows := conv.Rows()
 	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose())
 	rows := int64(0)
@@ -284,7 +285,7 @@ func dataFromSQL(config spanner.BatchWriterConfig, client *sp.Client, conv *inte
 		func(table string, cols []string, vals []interface{}) {
 			writer.AddRow(table, cols, vals)
 		})
-	internal.ProcessSqlData(conv, sourceDB)
+	postgres.ProcessSqlData(conv, sourceDB)
 	writer.Flush()
 	return writer, nil
 }
@@ -307,7 +308,7 @@ func schemaFromPgDump(ioHelper *ioStreams) (*internal.Conv, error) {
 	r := internal.NewReader(bufio.NewReader(f), p)
 	conv.SetSchemaMode() // Build schema and ignore data in pg_dump.
 	conv.SetDataSink(nil)
-	err = internal.ProcessPgDump(conv, r)
+	err = postgres.ProcessPgDump(conv, r)
 	if err != nil {
 		fmt.Fprintf(ioHelper.out, "Failed to parse the data file: %v", err)
 		return nil, fmt.Errorf("failed to parse the data file")
@@ -342,7 +343,7 @@ func dataFromPgDump(config spanner.BatchWriterConfig, ioHelper *ioStreams, clien
 		func(table string, cols []string, vals []interface{}) {
 			writer.AddRow(table, cols, vals)
 		})
-	internal.ProcessPgDump(conv, r)
+	postgres.ProcessPgDump(conv, r)
 	writer.Flush()
 	p.Done()
 
diff --git a/internal/data.go b/postgres/data.go
similarity index 96%
rename from internal/data.go
rename to postgres/data.go
index 758a199..7a9d1b4 100644
--- a/internal/data.go
+++ b/postgres/data.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"encoding/hex"
@@ -26,14 +26,15 @@ import (
 	"cloud.google.com/go/civil"
 	"cloud.google.com/go/spanner"
 
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
 
 // 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) {
+// to send to Spanner.  ProcessDataRow is only called in DataMode.
+func ProcessDataRow(conv *internal.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))
@@ -48,17 +49,17 @@ func ProcessDataRow(conv *Conv, srcTable string, srcCols, vals []string) {
 // 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) {
+func ConvertData(conv *internal.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)
+	spTable, err := internal.GetSpannerTable(conv, srcTable)
 	if err != nil {
 		return "", []string{}, []interface{}{}, fmt.Errorf("can't map source table %s", srcTable)
 	}
-	spCols, err := GetSpannerCols(conv, srcTable, srcCols)
+	spCols, err := internal.GetSpannerCols(conv, srcTable, srcCols)
 	if err != nil {
 		return "", []string{}, []interface{}{}, fmt.Errorf("can't map source columns %v", srcCols)
 	}
diff --git a/internal/infoschema.go b/postgres/infoschema.go
similarity index 92%
rename from internal/infoschema.go
rename to postgres/infoschema.go
index 09b8249..3dee9eb 100644
--- a/internal/infoschema.go
+++ b/postgres/infoschema.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"database/sql"
@@ -25,6 +25,7 @@ import (
 	"cloud.google.com/go/civil"
 	_ "github.com/lib/pq"
 
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
@@ -38,7 +39,7 @@ import (
 // 'db'. We assume that the source database supports information
 // schema tables. These tables are a broadly supported ANSI standard,
 // and we use them to obtain source database's schema information.
-func ProcessInfoSchema(conv *Conv, db *sql.DB) error {
+func ProcessInfoSchema(conv *internal.Conv, db *sql.DB) error {
 	tables, err := getTables(db)
 	if err != nil {
 		return err
@@ -72,7 +73,7 @@ func ProcessInfoSchema(conv *Conv, db *sql.DB) error {
 // We choose to do all type conversions explicitly ourselves so that
 // we can generate more targeted error messages: hence we pass
 // *interface{} parameters to row.Scan.
-func ProcessSqlData(conv *Conv, db *sql.DB) {
+func ProcessSqlData(conv *internal.Conv, db *sql.DB) {
 	// TODO: refactor to use the set of tables computed by
 	// ProcessInfoSchema instead of computing them again.
 	tables, err := getTables(db)
@@ -93,8 +94,8 @@ func ProcessSqlData(conv *Conv, db *sql.DB) {
 		defer rows.Close()
 		srcTable := buildTableName(t.schema, t.name)
 		srcCols, err1 := rows.Columns()
-		spTable, err2 := GetSpannerTable(conv, srcTable)
-		spCols, err3 := GetSpannerCols(conv, srcTable, srcCols)
+		spTable, err2 := internal.GetSpannerTable(conv, srcTable)
+		spCols, err3 := internal.GetSpannerCols(conv, srcTable, srcCols)
 		spSchema, ok1 := conv.SpSchema[spTable]
 		srcSchema, ok2 := conv.SrcSchema[srcTable]
 		if err1 != nil || err2 != nil || err3 != nil || !ok1 || !ok2 {
@@ -130,7 +131,7 @@ func ProcessSqlData(conv *Conv, db *sql.DB) {
 // ConvertSqlRow returns cols as well as converted values. This is
 // because cols can change when we add a column (synthetic primary
 // key) or because we drop columns (handling of NULL values).
-func ConvertSqlRow(conv *Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, srcVals []interface{}) ([]string, []interface{}, error) {
+func ConvertSqlRow(conv *internal.Conv, srcTable string, srcCols []string, srcSchema schema.Table, spTable string, spCols []string, spSchema ddl.CreateTable, srcVals []interface{}) ([]string, []interface{}, error) {
 	var vs []interface{}
 	var cs []string
 	for i := range srcCols {
@@ -165,7 +166,7 @@ func ConvertSqlRow(conv *Conv, srcTable string, srcCols []string, srcSchema sche
 }
 
 // SetRowStats populates conv with the number of rows in each table.
-func SetRowStats(conv *Conv, db *sql.DB) {
+func SetRowStats(conv *internal.Conv, db *sql.DB) {
 	// TODO: refactor to use the set of tables computed by
 	// ProcessInfoSchema instead of computing them again.
 	tables, err := getTables(db)
@@ -225,7 +226,7 @@ func getTables(db *sql.DB) ([]schemaAndName, error) {
 	return tables, nil
 }
 
-func processTable(conv *Conv, db *sql.DB, table schemaAndName) error {
+func processTable(conv *internal.Conv, db *sql.DB, table schemaAndName) error {
 	cols, err := getColumns(table, db)
 	if err != nil {
 		return fmt.Errorf("couldn't get schema for table %s.%s: %s\n", table.schema, table.name, err)
@@ -258,7 +259,7 @@ func getColumns(table schemaAndName, db *sql.DB) (*sql.Rows, error) {
 	return db.Query(q, table.schema, table.name)
 }
 
-func processColumns(conv *Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string) {
+func processColumns(conv *internal.Conv, cols *sql.Rows, constraints map[string][]string) (map[string]schema.Column, []string) {
 	colDefs := make(map[string]schema.Column)
 	var colNames []string
 	var colName, dataType, isNullable string
@@ -302,7 +303,7 @@ func processColumns(conv *Conv, cols *sql.Rows, constraints map[string][]string)
 // getConstraints returns a list of primary keys and by-column map of
 // other constraints.  Note: we need to preserve ordinal order of
 // columns in primary key constraints.
-func getConstraints(conv *Conv, db *sql.DB, table schemaAndName) ([]string, map[string][]string, error) {
+func getConstraints(conv *internal.Conv, db *sql.DB, table schemaAndName) ([]string, map[string][]string, error) {
 	q := `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
               FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
                 INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
@@ -353,7 +354,7 @@ func toType(dataType string, elementDataType sql.NullString, charLen sql.NullInt
 	}
 }
 
-func toNotNull(conv *Conv, isNullable string) bool {
+func toNotNull(conv *internal.Conv, isNullable string) bool {
 	switch isNullable {
 	case "YES":
 		return false
@@ -364,7 +365,7 @@ func toNotNull(conv *Conv, isNullable string) bool {
 	return false
 }
 
-func cvtSqlArray(conv *Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interface{}) (interface{}, error) {
+func cvtSqlArray(conv *internal.Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interface{}) (interface{}, error) {
 	a, ok := val.([]byte)
 	if !ok {
 		return nil, fmt.Errorf("can't convert array values to []byte")
@@ -386,7 +387,7 @@ func cvtSqlArray(conv *Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interf
 //    float64
 //    string
 //    time.Time
-func cvtSqlScalar(conv *Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interface{}) (interface{}, error) {
+func cvtSqlScalar(conv *internal.Conv, srcCd schema.Column, spCd ddl.ColumnDef, val interface{}) (interface{}, error) {
 	switch spCd.T.(type) {
 	case ddl.Bool:
 		switch v := val.(type) {
diff --git a/internal/pgdump.go b/postgres/pgdump.go
similarity index 89%
rename from internal/pgdump.go
rename to postgres/pgdump.go
index 21951c4..623070c 100644
--- a/internal/pgdump.go
+++ b/postgres/pgdump.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"fmt"
@@ -24,6 +24,7 @@ import (
 	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/schema"
 )
 
@@ -46,7 +47,7 @@ const (
 // 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 {
+func ProcessPgDump(conv *internal.Conv, r *internal.Reader) error {
 	for {
 		startLine := r.LineNumber
 		startOffset := r.Offset
@@ -55,7 +56,7 @@ func ProcessPgDump(conv *Conv, r *Reader) error {
 			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)
+		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:
@@ -78,7 +79,7 @@ func ProcessPgDump(conv *Conv, r *Reader) error {
 
 // 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) {
+func readAndParseChunk(conv *internal.Conv, r *internal.Reader) ([]byte, []nodes.Node, error) {
 	var l [][]byte
 	for {
 		b := r.ReadLine()
@@ -112,12 +113,12 @@ func readAndParseChunk(conv *Conv, r *Reader) ([]byte, []nodes.Node, error) {
 	}
 }
 
-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)
+func processCopyBlock(conv *internal.Conv, srcTable string, srcCols []string, 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" {
-			VerbosePrintf("Parsed COPY-FROM stdin block ending at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
+			internal.VerbosePrintf("Parsed COPY-FROM stdin block ending at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
 			return
 		}
 		if r.EOF {
@@ -149,7 +150,7 @@ func processCopyBlock(conv *Conv, srcTable string, srcCols []string, r *Reader)
 // copyOrInsert if a COPY-FROM or INSERT statement is encountered.
 // Note that the actual parsing/processing of COPY-FROM data blocks is
 // handled elsewhere (see process.go).
-func processStatements(conv *Conv, statements []nodes.Node) *copyOrInsert {
+func processStatements(conv *internal.Conv, statements []nodes.Node) *copyOrInsert {
 	// Typically we'll have only one statement, but we handle the general case.
 	for i, node := range statements {
 		switch n := node.(type) {
@@ -185,7 +186,7 @@ func processStatements(conv *Conv, statements []nodes.Node) *copyOrInsert {
 	return nil
 }
 
-func processAlterTableStmt(conv *Conv, n nodes.AlterTableStmt) {
+func processAlterTableStmt(conv *internal.Conv, n nodes.AlterTableStmt) {
 	if n.Relation == nil {
 		logStmtError(conv, n, fmt.Errorf("relation is nil"))
 		return
@@ -226,11 +227,11 @@ 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(prNodes([]nodes.Node{n}))
-		VerbosePrintf("Processing %v statement: table %s not found", reflect.TypeOf(n), table)
+		internal.VerbosePrintf("Processing %v statement: table %s not found", reflect.TypeOf(n), table)
 	}
 }
 
-func processCreateStmt(conv *Conv, n nodes.CreateStmt) {
+func processCreateStmt(conv *internal.Conv, n nodes.CreateStmt) {
 	var colNames []string
 	colDef := make(map[string]schema.Column)
 	if n.Relation == nil {
@@ -272,7 +273,7 @@ func processCreateStmt(conv *Conv, n nodes.CreateStmt) {
 	updateSchema(conv, table, constraints, "CREATE TABLE")
 }
 
-func processColumn(conv *Conv, n nodes.ColumnDef, table string) (string, schema.Column, []constraint, error) {
+func processColumn(conv *internal.Conv, n nodes.ColumnDef, table string) (string, schema.Column, []constraint, error) {
 	mods := getTypeMods(conv, n.TypeName.Typmods)
 	if n.Colname == nil {
 		return "", schema.Column{}, nil, fmt.Errorf("colname is nil")
@@ -289,7 +290,7 @@ func processColumn(conv *Conv, n nodes.ColumnDef, table string) (string, schema.
 	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 {
+func processInsertStmt(conv *internal.Conv, n nodes.InsertStmt) *copyOrInsert {
 	if n.Relation == nil {
 		logStmtError(conv, n, fmt.Errorf("relation is nil"))
 		return nil
@@ -320,7 +321,7 @@ func processInsertStmt(conv *Conv, n nodes.InsertStmt) *copyOrInsert {
 	return nil
 }
 
-func processCopyStmt(conv *Conv, n nodes.CopyStmt) *copyOrInsert {
+func processCopyStmt(conv *internal.Conv, n nodes.CopyStmt) *copyOrInsert {
 	// Always return a copyOrInsert{stmt: copyFrom, ...} even if we
 	// encounter errors. Otherwise we won't be able to parse
 	// the data portion of the COPY-FROM statement, and we'll
@@ -348,7 +349,7 @@ func processCopyStmt(conv *Conv, n nodes.CopyStmt) *copyOrInsert {
 	return &copyOrInsert{stmt: copyFrom, table: table, cols: cols}
 }
 
-func processVariableSetStmt(conv *Conv, n nodes.VariableSetStmt) {
+func processVariableSetStmt(conv *internal.Conv, n nodes.VariableSetStmt) {
 	if n.Name != nil && *n.Name == "timezone" {
 		if len(n.Args.Items) == 1 {
 			switch c := n.Args.Items[0].(type) {
@@ -372,7 +373,7 @@ func processVariableSetStmt(conv *Conv, n nodes.VariableSetStmt) {
 	}
 }
 
-func getTypeMods(conv *Conv, t nodes.List) (l []int64) {
+func getTypeMods(conv *internal.Conv, t nodes.List) (l []int64) {
 	for _, x := range t.Items {
 		switch t1 := x.(type) {
 		case nodes.A_Const:
@@ -389,7 +390,7 @@ func getTypeMods(conv *Conv, t nodes.List) (l []int64) {
 	return l
 }
 
-func getArrayBounds(conv *Conv, t nodes.List) (l []int64) {
+func getArrayBounds(conv *internal.Conv, t nodes.List) (l []int64) {
 	for _, x := range t.Items {
 		switch t := x.(type) {
 		case nodes.Integer:
@@ -425,7 +426,7 @@ func getTypeID(l []nodes.Node) (string, error) {
 
 // getTableName extracts the table name from RangeVar n, and returns
 // the raw extracted name (the PostgreSQL table name).
-func getTableName(conv *Conv, n nodes.RangeVar) (string, error) {
+func getTableName(conv *internal.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"
@@ -459,7 +460,7 @@ type constraint struct {
 // extractConstraints traverses a list of nodes (expecting them to be
 // Constraint nodes), and collects the contraints they represent as
 // a list of constraint-type/column-names pairs.
-func extractConstraints(conv *Conv, n nodes.Node, table string, l []nodes.Node) (cs []constraint) {
+func extractConstraints(conv *internal.Conv, n nodes.Node, table string, l []nodes.Node) (cs []constraint) {
 	for _, i := range l {
 		switch d := i.(type) {
 		case nodes.Constraint:
@@ -485,7 +486,7 @@ func extractConstraints(conv *Conv, n nodes.Node, table 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, table string, l []nodes.Node, pgCol string) (cs []constraint) {
+func analyzeColDefConstraints(conv *internal.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, table, l) {
@@ -500,7 +501,7 @@ func analyzeColDefConstraints(conv *Conv, n nodes.Node, table string, l []nodes.
 
 // 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) {
+func updateSchema(conv *internal.Conv, table string, cs []constraint, s string) {
 	for _, c := range cs {
 		switch c.ct {
 		case nodes.CONSTR_PRIMARY:
@@ -542,7 +543,7 @@ func updateCols(ct nodes.ConstrType, colNames []string, colDef map[string]schema
 
 // toSchemaKeys converts a string list of PostgreSQL primary keys to
 // schema primary keys.
-func toSchemaKeys(conv *Conv, table string, s []string) (l []schema.Key) {
+func toSchemaKeys(conv *internal.Conv, table string, s []string) (l []schema.Key) {
 	for _, k := range s {
 		// PostgreSQL primary keys have no notation of ascending/descending.
 		// We map them all into ascending primarary keys.
@@ -552,7 +553,7 @@ func toSchemaKeys(conv *Conv, table string, s []string) (l []schema.Key) {
 }
 
 // getCols extracts and returns the column names for an InsertStatement.
-func getCols(conv *Conv, table string, l []nodes.Node) (cols []string, err error) {
+func getCols(conv *internal.Conv, table string, l []nodes.Node) (cols []string, err error) {
 	for _, n := range l {
 		switch r := n.(type) {
 		case nodes.ResTarget:
@@ -567,7 +568,7 @@ func getCols(conv *Conv, table string, l []nodes.Node) (cols []string, err error
 }
 
 // getVals extracts and returns the values for an InsertStatement.
-func getVals(conv *Conv, l [][]nodes.Node, n nodes.InsertStmt) (values []string) {
+func getVals(conv *internal.Conv, l [][]nodes.Node, n nodes.InsertStmt) (values []string) {
 	for _, vl := range l {
 		for _, v := range vl {
 			switch c := v.(type) {
@@ -592,7 +593,7 @@ func getVals(conv *Conv, l [][]nodes.Node, n nodes.InsertStmt) (values []string)
 	return values
 }
 
-func logStmtError(conv *Conv, n nodes.Node, err error) {
+func logStmtError(conv *internal.Conv, n nodes.Node, err error) {
 	conv.Unexpected(fmt.Sprintf("Processing %v statement: %s", reflect.TypeOf(n), err))
 	conv.ErrorInStatement(prNodes([]nodes.Node{n}))
 }
@@ -608,7 +609,7 @@ 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 []schema.Key, s string) {
+func checkEmpty(conv *internal.Conv, pkeys []schema.Key, s string) {
 	if len(pkeys) != 0 {
 		conv.Unexpected(fmt.Sprintf("%s statement is adding a second primary key", s))
 	}
diff --git a/internal/toddl.go b/postgres/toddl.go
similarity index 75%
rename from internal/toddl.go
rename to postgres/toddl.go
index 420f7db..1205bb0 100644
--- a/internal/toddl.go
+++ b/postgres/toddl.go
@@ -12,14 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"fmt"
 	"strconv"
-	"strings"
 	"unicode"
 
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
@@ -27,20 +27,20 @@ import (
 // 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 {
+func schemaToDDL(conv *internal.Conv) error {
 	for _, srcTable := range conv.SrcSchema {
-		spTableName, err := GetSpannerTable(conv, srcTable.Name)
+		spTableName, err := internal.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)
+		conv.Issues[srcTable.Name] = make(map[string][]internal.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)
+			colName, err := internal.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
@@ -49,14 +49,14 @@ func schemaToDDL(conv *Conv) error {
 			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)
+				issues = append(issues, internal.MultiDimensionalArray)
 			}
 			// TODO: add issues for all elements of srcCol.Ignored.
 			if srcCol.Ignored.ForeignKey {
-				issues = append(issues, ForeignKey)
+				issues = append(issues, internal.ForeignKey)
 			}
 			if srcCol.Ignored.Default {
-				issues = append(issues, DefaultValue)
+				issues = append(issues, internal.DefaultValue)
 			}
 			if len(issues) > 0 {
 				conv.Issues[srcTable.Name][srcCol.Name] = issues
@@ -66,7 +66,7 @@ func schemaToDDL(conv *Conv) error {
 				T:       ty,
 				IsArray: len(srcCol.Type.ArrayBounds) == 1,
 				NotNull: srcCol.NotNull,
-				Comment: "From: " + quoteIfNeeded(srcCol.Name) + " " + printSourceType(srcCol.Type),
+				Comment: "From: " + quoteIfNeeded(srcCol.Name) + " " + srcCol.Type.Print(),
 			}
 		}
 		comment := "Spanner schema for source table " + quoteIfNeeded(srcTable.Name)
@@ -84,7 +84,7 @@ func schemaToDDL(conv *Conv) error {
 // 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) {
+func toSpannerType(conv *internal.Conv, id string, mods []int64) (ddl.ScalarType, []internal.SchemaIssue) {
 	maxExpectedMods := func(n int) {
 		if len(mods) > n {
 			conv.Unexpected(fmt.Sprintf("Found %d mods while processing type id=%s", len(mods), id))
@@ -96,7 +96,7 @@ func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []Schem
 		return ddl.Bool{}, nil
 	case "bigserial":
 		maxExpectedMods(0)
-		return ddl.Int64{}, []SchemaIssue{Serial}
+		return ddl.Int64{}, []internal.SchemaIssue{internal.Serial}
 	case "bpchar", "character": // Note: Postgres internal name for char is bpchar (aka blank padded char).
 		maxExpectedMods(1)
 		if len(mods) > 0 {
@@ -115,28 +115,28 @@ func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []Schem
 		return ddl.Float64{}, nil
 	case "float4", "real":
 		maxExpectedMods(0)
-		return ddl.Float64{}, []SchemaIssue{Widened}
+		return ddl.Float64{}, []internal.SchemaIssue{internal.Widened}
 	case "int8", "bigint":
 		maxExpectedMods(0)
 		return ddl.Int64{}, nil
 	case "int4", "integer":
 		maxExpectedMods(0)
-		return ddl.Int64{}, []SchemaIssue{Widened}
+		return ddl.Int64{}, []internal.SchemaIssue{internal.Widened}
 	case "int2", "smallint":
 		maxExpectedMods(0)
-		return ddl.Int64{}, []SchemaIssue{Widened}
+		return ddl.Int64{}, []internal.SchemaIssue{internal.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{}, []internal.SchemaIssue{internal.NumericThatFits}
 		}
-		return ddl.Float64{}, []SchemaIssue{Numeric}
+		return ddl.Float64{}, []internal.SchemaIssue{internal.Numeric}
 	case "serial":
 		maxExpectedMods(0)
-		return ddl.Int64{}, []SchemaIssue{Serial}
+		return ddl.Int64{}, []internal.SchemaIssue{internal.Serial}
 	case "text":
 		maxExpectedMods(0)
 		return ddl.String{Len: ddl.MaxLength{}}, nil
@@ -146,7 +146,7 @@ func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []Schem
 	case "timestamp", "timestamp without time zone":
 		maxExpectedMods(1)
 		// Map timestamp without timezone to Spanner timestamp.
-		return ddl.Timestamp{}, []SchemaIssue{Timestamp}
+		return ddl.Timestamp{}, []internal.SchemaIssue{internal.Timestamp}
 	case "varchar", "character varying":
 		maxExpectedMods(1)
 		if len(mods) > 0 {
@@ -154,30 +154,7 @@ func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []Schem
 		}
 		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
+	return ddl.String{Len: ddl.MaxLength{}}, []internal.SchemaIssue{internal.NoGoodType}
 }
 
 func quoteIfNeeded(s string) string {
@@ -190,10 +167,10 @@ func quoteIfNeeded(s string) string {
 	return s
 }
 
-func cvtPrimaryKeys(conv *Conv, srcTable string, srcKeys []schema.Key) []ddl.IndexKey {
+func cvtPrimaryKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Key) []ddl.IndexKey {
 	var spKeys []ddl.IndexKey
 	for _, k := range srcKeys {
-		spCol, err := GetSpannerCol(conv, srcTable, k.Column, true)
+		spCol, err := internal.GetSpannerCol(conv, srcTable, k.Column, true)
 		if err != nil {
 			conv.Unexpected(fmt.Sprintf("Can't map key for table %s", srcTable))
 			continue
diff --git a/schema/schema.go b/schema/schema.go
index e7d9219..b6dd7cc 100644
--- a/schema/schema.go
+++ b/schema/schema.go
@@ -25,6 +25,12 @@
 // schema, and potentially get rid of the ddl package.
 package schema
 
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
 // Table represents a database table.
 type Table struct {
 	Name        string
@@ -73,3 +79,26 @@ type Ignored struct {
 	Exclusion  bool
 	ForeignKey bool
 }
+
+func (ty Type) Print() 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
+}


Test patch:
diff --git a/internal/convert_test.go b/internal/convert_test.go
index d811277..964b6a4 100644
--- a/internal/convert_test.go
+++ b/internal/convert_test.go
@@ -90,16 +90,16 @@ func TestBadRows(t *testing.T) {
 
 func TestStatements(t *testing.T) {
 	conv := MakeConv()
-	conv.ErrorInStatement(prNodes(parse(t, "CREATE TABLE cart (pid text NOT NULL);")))
-	conv.SchemaStatement(prNodes(parse(t, "CREATE TABLE cart (pid text NOT NULL);")))
-	conv.DataStatement(prNodes(parse(t, "INSERT INTO cart (pid) VALUES ('p42');")))
-	conv.SkipStatement(prNodes(parse(t, "GRANT ALL ON SCHEMA public TO PUBLIC;")))
+	conv.ErrorInStatement("Error statement")
+	conv.SchemaStatement("Schema statement")
+	conv.DataStatement("Data statement")
+	conv.SkipStatement("Skip statement")
 	assert.Equal(t, int64(4), conv.Statements())
 }
 
 func TestStatementErrors(t *testing.T) {
 	conv := MakeConv()
-	conv.ErrorInStatement(prNodes(parse(t, "CREATE TABLE cart (pid text NOT NULL);")))
+	conv.ErrorInStatement("Error statement")
 	assert.Equal(t, int64(1), conv.StatementErrors())
 }
 
diff --git a/internal/data_test.go b/postgres/data_test.go
similarity index 96%
rename from internal/data_test.go
rename to postgres/data_test.go
index a487b97..3648508 100644
--- a/internal/data_test.go
+++ b/postgres/data_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"fmt"
@@ -22,10 +22,10 @@ import (
 
 	"cloud.google.com/go/civil"
 	"cloud.google.com/go/spanner"
-	"github.com/stretchr/testify/assert"
-
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	"github.com/stretchr/testify/assert"
 )
 
 // Basic smoke test of ProcessDataRow. The core part of this code path
@@ -270,19 +270,19 @@ func TestConvertData(t *testing.T) {
 		},
 	}
 	conv := buildConv(spTable, srcTable)
-	conv.SyntheticPKeys[spTable.Name] = SyntheticPKey{Col: "synth_id", Sequence: 0}
+	conv.SyntheticPKeys[spTable.Name] = internal.SyntheticPKey{Col: "synth_id", Sequence: 0}
 	for _, tc := range syntheticPKeyTests {
 		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(spTable ddl.CreateTable, srcTable schema.Table) *Conv {
-	conv := MakeConv()
+func buildConv(spTable ddl.CreateTable, srcTable schema.Table) *internal.Conv {
+	conv := internal.MakeConv()
 	conv.SpSchema[spTable.Name] = spTable
 	conv.SrcSchema[srcTable.Name] = srcTable
-	conv.ToSource[spTable.Name] = NameAndCols{Name: srcTable.Name, Cols: make(map[string]string)}
-	conv.ToSpanner[srcTable.Name] = NameAndCols{Name: spTable.Name, Cols: make(map[string]string)}
+	conv.ToSource[spTable.Name] = internal.NameAndCols{Name: srcTable.Name, Cols: make(map[string]string)}
+	conv.ToSpanner[srcTable.Name] = internal.NameAndCols{Name: spTable.Name, Cols: make(map[string]string)}
 	for i := range spTable.ColNames {
 		conv.ToSource[spTable.Name].Cols[spTable.ColNames[i]] = srcTable.ColNames[i]
 		conv.ToSpanner[srcTable.Name].Cols[srcTable.ColNames[i]] = spTable.ColNames[i]
diff --git a/internal/infoschema_test.go b/postgres/infoschema_test.go
similarity index 96%
rename from internal/infoschema_test.go
rename to postgres/infoschema_test.go
index 48589b8..0d11e29 100644
--- a/internal/infoschema_test.go
+++ b/postgres/infoschema_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"database/sql"
@@ -22,6 +22,7 @@ import (
 
 	"cloud.google.com/go/spanner"
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
@@ -92,7 +93,7 @@ func TestProcessInfoSchema(t *testing.T) {
 		},
 	}
 	db := mkMockDB(t, ms)
-	conv := MakeConv()
+	conv := internal.MakeConv()
 	err := ProcessInfoSchema(conv, db)
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
@@ -135,15 +136,15 @@ func TestProcessInfoSchema(t *testing.T) {
 	}
 	assert.Equal(t, expectedSchema, stripSchemaComments(conv.SpSchema))
 	assert.Equal(t, len(conv.Issues["cart"]), 0)
-	expectedIssues := map[string][]SchemaIssue{
-		"aint": []SchemaIssue{Widened},
-		"bs":   []SchemaIssue{DefaultValue},
-		"f4":   []SchemaIssue{Widened},
-		"i4":   []SchemaIssue{Widened},
-		"i2":   []SchemaIssue{Widened},
-		"num":  []SchemaIssue{Numeric},
-		"s":    []SchemaIssue{Widened, DefaultValue},
-		"ts":   []SchemaIssue{Timestamp},
+	expectedIssues := map[string][]internal.SchemaIssue{
+		"aint": []internal.SchemaIssue{internal.Widened},
+		"bs":   []internal.SchemaIssue{internal.DefaultValue},
+		"f4":   []internal.SchemaIssue{internal.Widened},
+		"i4":   []internal.SchemaIssue{internal.Widened},
+		"i2":   []internal.SchemaIssue{internal.Widened},
+		"num":  []internal.SchemaIssue{internal.Numeric},
+		"s":    []internal.SchemaIssue{internal.Widened, internal.DefaultValue},
+		"ts":   []internal.SchemaIssue{internal.Timestamp},
 	}
 	assert.Equal(t, expectedIssues, conv.Issues["test"])
 	assert.Equal(t, int64(0), conv.Unexpecteds())
@@ -193,6 +194,7 @@ func TestProcessSqlData(t *testing.T) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
 	ProcessSqlData(conv, db)
+
 	assert.Equal(t,
 		[]spannerData{
 			spannerData{table: "te_st", cols: []string{"a a", " b", " c "}, vals: []interface{}{float64(42.3), int64(3), "cat"}},
@@ -260,7 +262,7 @@ func TestConvertSqlRow_SingleCol(t *testing.T) {
 	tableName := "testtable"
 	for _, tc := range tc {
 		col := "a"
-		conv := MakeConv()
+		conv := internal.MakeConv()
 		conv.SetLocation(time.UTC)
 		cols := []string{col}
 		srcSchema := schema.Table{Name: tableName, ColNames: []string{col}, ColDefs: map[string]schema.Column{col: schema.Column{Type: tc.srcType}}}
@@ -318,7 +320,7 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 		},
 	}
 	db := mkMockDB(t, ms)
-	conv := MakeConv()
+	conv := internal.MakeConv()
 	err := ProcessInfoSchema(conv, db)
 	assert.Nil(t, err)
 	conv.SetDataMode()
@@ -352,7 +354,7 @@ func TestSetRowStats(t *testing.T) {
 		},
 	}
 	db := mkMockDB(t, ms)
-	conv := MakeConv()
+	conv := internal.MakeConv()
 	conv.SetDataMode()
 	SetRowStats(conv, db)
 	assert.Equal(t, int64(5), conv.Stats.Rows["test1"])
diff --git a/internal/pgdump_test.go b/postgres/pgdump_test.go
similarity index 98%
rename from internal/pgdump_test.go
rename to postgres/pgdump_test.go
index 04219db..fb3b500 100644
--- a/internal/pgdump_test.go
+++ b/postgres/pgdump_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"bufio"
@@ -23,10 +23,11 @@ import (
 	"time"
 
 	"cloud.google.com/go/spanner"
-	pg_query "github.com/lfittl/pg_query_go"
-	"github.com/stretchr/testify/assert"
 
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+	pg_query "github.com/lfittl/pg_query_go"
+	"github.com/stretchr/testify/assert"
 )
 
 type spannerData struct {
@@ -547,10 +548,10 @@ func TestProcessPgDump_AddPrimaryKeys(t *testing.T) {
 
 func TestProcessPgDump_WithUnparsableContent(t *testing.T) {
 	s := "This is unparsable content"
-	conv := MakeConv()
+	conv := internal.MakeConv()
 	conv.SetLocation(time.UTC)
 	conv.SetSchemaMode()
-	err := ProcessPgDump(conv, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	err := ProcessPgDump(conv, internal.NewReader(bufio.NewReader(strings.NewReader(s)), nil))
 	if err == nil {
 		t.Fatalf("Expect an error, but got nil")
 	}
@@ -559,18 +560,18 @@ func TestProcessPgDump_WithUnparsableContent(t *testing.T) {
 	}
 }
 
-func runProcessPgDump(s string) (*Conv, []spannerData) {
-	conv := MakeConv()
+func runProcessPgDump(s string) (*internal.Conv, []spannerData) {
+	conv := internal.MakeConv()
 	conv.SetLocation(time.UTC)
 	conv.SetSchemaMode()
-	ProcessPgDump(conv, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	ProcessPgDump(conv, internal.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, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	ProcessPgDump(conv, internal.NewReader(bufio.NewReader(strings.NewReader(s)), nil))
 	return conv, rows
 }
 
@@ -578,7 +579,7 @@ func runProcessPgDump(s string) (*Conv, []spannerData) {
 // contains no unexpected conditions, statement errors, etc. Note that
 // many tests are issue-free, but several explicitly test handling of
 // various issues (so don't call nonIssue for them!).
-func noIssues(conv *Conv, t *testing.T, name string) {
+func noIssues(conv *internal.Conv, t *testing.T, name string) {
 	assert.Zero(t, len(conv.Stats.Unexpected), fmt.Sprintf("'%s' generated unexpected conditions: %v", name, conv.Stats.Unexpected))
 	for s, stat := range conv.Stats.Statement {
 		assert.Zero(t, stat.Error, fmt.Sprintf("'%s' generated %d errors for %s statements", name, stat.Error, s))
diff --git a/internal/report_test.go b/postgres/report_test.go
similarity index 96%
rename from internal/report_test.go
rename to postgres/report_test.go
index 2ded744..0b93739 100644
--- a/internal/report_test.go
+++ b/postgres/report_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"bufio"
@@ -21,6 +21,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -44,9 +45,9 @@ func TestReport(t *testing.T) {
             a bigint[],
             b integer NOT NULL,
             c text);`
-	conv := MakeConv()
+	conv := internal.MakeConv()
 	conv.SetSchemaMode()
-	ProcessPgDump(conv, NewReader(bufio.NewReader(strings.NewReader(s)), nil))
+	ProcessPgDump(conv, internal.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}
@@ -54,7 +55,7 @@ func TestReport(t *testing.T) {
 	conv.Stats.Unexpected["Testing unexpected messages"] = 5
 	buf := new(bytes.Buffer)
 	w := bufio.NewWriter(buf)
-	GenerateReport(true, conv, w, badWrites)
+	internal.GenerateReport(true, conv, w, badWrites)
 	w.Flush()
 	// Print copy of report to stdout (shows up when running go test -v).
 	fmt.Print(buf.String())
diff --git a/internal/toddl_test.go b/postgres/toddl_test.go
similarity index 90%
rename from internal/toddl_test.go
rename to postgres/toddl_test.go
index 5e133a8..c05513e 100644
--- a/internal/toddl_test.go
+++ b/postgres/toddl_test.go
@@ -12,11 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package postgres
 
 import (
 	"testing"
 
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/stretchr/testify/assert"
@@ -26,7 +27,7 @@ import (
 // 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 := internal.MakeConv()
 	conv.SetSchemaMode()
 	name := "test"
 	srcSchema := schema.Table{
@@ -59,9 +60,9 @@ func TestToSpannerType(t *testing.T) {
 		Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a"}},
 	}
 	assert.Equal(t, expected, actual)
-	expectedIssues := map[string][]SchemaIssue{
-		"b": []SchemaIssue{Widened},
-		"e": []SchemaIssue{Numeric},
+	expectedIssues := map[string][]internal.SchemaIssue{
+		"b": []internal.SchemaIssue{internal.Widened},
+		"e": []internal.SchemaIssue{internal.Numeric},
 	}
 	assert.Equal(t, expectedIssues, conv.Issues[name])
 }