Repository: cloudspannerecosystem/harbourbridge
Pull request #50 (2020-03-02 05:23:46)

Add initial version of infoschema processing for PostgreSQL.
As an alternative to processing pg_dump data, we're adding support for
accessing database schema information via information_schema
tables. This is our first step towards adding support for databases
other that PostgreSQL.

This initial PR:
a) adds the core support for accessing information_schema tables (just
   tables/columns and their types and constraints)
b) hooks this new path into main.go via a new 'driver' flag

Follow-up PRs will:
- add sql queries for accessing functions, procedures, views,
  triggers, indexes.
- add support for extracting data from source DB, converting data and
  inserting in Spanner
- add support for other DBs e.g. mysql


Patch:
diff --git a/go.mod b/go.mod
index 5690a09..bbeb5dc 100644
--- a/go.mod
+++ b/go.mod
@@ -7,8 +7,10 @@ require (
 	cloud.google.com/go/spanner v1.0.1-0.20191018063856-b24632667359
 	github.com/goware/modvendor v0.0.0-20190516042800-ce72b408a8fe // indirect
 	github.com/lfittl/pg_query_go v1.0.0
+	github.com/lib/pq v1.3.0
 	github.com/mattn/go-zglob v0.0.1 // indirect
 	github.com/stretchr/testify v1.4.0
+	golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550
 	golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f // indirect
 	golang.org/x/tools v0.0.0-20200125223703-d33eef8e6825 // indirect
 	google.golang.org/api v0.9.0
diff --git a/go.sum b/go.sum
index 84ba8c7..d6f3735 100644
--- a/go.sum
+++ b/go.sum
@@ -55,6 +55,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lfittl/pg_query_go v1.0.0 h1:rcHZK5DBEUoxtO6dACP+UVCHKtA1ZsELBW0rSjOXMAE=
 github.com/lfittl/pg_query_go v1.0.0/go.mod h1:jcikG62RKf+NIWmbLzjjk73m4x6um2pKf3h+TJyINms=
+github.com/lib/pq v1.3.0 h1:/qkRGz8zljWiDcFvgpwUpwIAPu3r07TDvs3Rws+o/pU=
+github.com/lib/pq v1.3.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/mattn/go-zglob v0.0.0-20180803001819-2ea3427bfa53 h1:tGfIHhDghvEnneeRhODvGYOt305TPwingKt6p90F4MU=
 github.com/mattn/go-zglob v0.0.0-20180803001819-2ea3427bfa53/go.mod h1:9fxibJccNxU2cnpIKLRRFA7zX7qhkJIQWBb449FYHOo=
 github.com/mattn/go-zglob v0.0.1 h1:xsEx/XUoVlI6yXjqBK062zYhRTZltCNmYPx6v+8DNaY=
@@ -71,6 +73,7 @@ go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550 h1:ObdrDkeb4kJdCP557AjRjq69pTHfNouLtWZG7j9rPN8=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
diff --git a/internal/infoschema.go b/internal/infoschema.go
new file mode 100644
index 0000000..41d68d2
--- /dev/null
+++ b/internal/infoschema.go
@@ -0,0 +1,247 @@
+// 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 (
+	"database/sql"
+	"fmt"
+	"os"
+	"strings"
+	"syscall"
+
+	_ "github.com/lib/pq"
+	"golang.org/x/crypto/ssh/terminal"
+
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+)
+
+// ProcessInfoSchema performs schema conversion for source database.
+// Schema information is obtained from the information_schema tables.
+// The driver used to access the database is specified by 'driver'.
+func ProcessInfoSchema(conv *Conv, driver string) error {
+	server := os.Getenv("PGHOST")
+	port := os.Getenv("PGPORT")
+	user := os.Getenv("PGUSER")
+	dbname := os.Getenv("PGDATABASE")
+	if server == "" || port == "" || user == "" || dbname == "" {
+		fmt.Printf("Please specify host, port, user and database using PGHOST, PGPORT, PGUSER and PGDATABASE environment variables\n")
+		return fmt.Errorf("Could not connect to source database")
+	}
+	password := getPassword()
+	db, err := sql.Open(driver, fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbname))
+	if err != nil {
+		return err
+	}
+	// TODO: All of the queries to get tables and table data
+	// should be in a single transaction to ensure we obtain a
+	// consistent snapshot of schema information across tables
+	// (pg_dump does something similar). When we add SELECT
+	// queries to get data, we should wrap those in the same
+	// transaction to ensure consistency of schema and data.
+	tables, err := getTables(db)
+	if err != nil {
+		return err
+	}
+	for _, t := range tables {
+		processTable(conv, db, t)
+	}
+	schemaToDDL(conv)
+	conv.AddPrimaryKeys()
+	return nil
+}
+
+func getPassword() string {
+	password := os.Getenv("PGPASSWORD")
+	if password != "" {
+		return password
+	}
+	fmt.Print("Enter Password: ")
+	bytePassword, err := terminal.ReadPassword(int(syscall.Stdin))
+	if err != nil {
+		fmt.Println("\nCoudln't read password")
+		return ""
+	}
+	fmt.Printf("\n")
+	return strings.TrimSpace(string(bytePassword))
+}
+
+type schemaAndName struct {
+	schema string
+	name   string
+}
+
+func getTables(db *sql.DB) ([]schemaAndName, error) {
+	ignoredSchemas := make(map[string]bool)
+	for _, s := range []string{"information_schema", "postgres", "pg_catalog", "pg_temp_1", "pg_toast", "pg_toast_temp_1"} {
+		ignoredSchemas[s] = true
+	}
+	q := "SELECT table_schema, table_name FROM information_schema.tables where table_type = 'BASE TABLE'"
+	rows, err := db.Query(q)
+	if err != nil {
+		return nil, fmt.Errorf("Couldn't get tables: %w\n", err)
+	}
+	defer rows.Close()
+	var tableSchema, tableName string
+	var tables []schemaAndName
+	for rows.Next() {
+		rows.Scan(&tableSchema, &tableName)
+		if !ignoredSchemas[tableSchema] {
+			tables = append(tables, schemaAndName{schema: tableSchema, name: tableName})
+		}
+	}
+	return tables, nil
+}
+
+func processTable(conv *Conv, db *sql.DB, table schemaAndName) error {
+	cols, err := getColumns(table, db)
+	if err != nil {
+		return fmt.Errorf("Couldn't get schema for table %s.%s: %s\n", table.schema, table.name, err)
+	}
+	defer cols.Close()
+	primaryKeys, constraints, err := getConstraints(conv, db, table)
+	if err != nil {
+		return fmt.Errorf("Couldn't get constraints for table %s.%s: %s\n", table.schema, table.name, err)
+	}
+	colDefs, colNames := processColumns(conv, cols, constraints)
+	name := fmt.Sprintf("%s.%s", table.schema, table.name)
+	if table.schema == "public" { // Drop 'public' prefix.
+		name = table.name
+	}
+	var schemaPKeys []schema.Key
+	for _, k := range primaryKeys {
+		schemaPKeys = append(schemaPKeys, schema.Key{Column: k})
+	}
+	conv.srcSchema[name] = schema.Table{
+		Name:        name,
+		ColNames:    colNames,
+		ColDefs:     colDefs,
+		PrimaryKeys: schemaPKeys}
+	return nil
+}
+
+func getColumns(table schemaAndName, db *sql.DB) (*sql.Rows, error) {
+	q := `SELECT c.column_name, c.data_type, e.data_type, c.is_nullable, c.column_default, c.character_maximum_length, c.numeric_precision, c.numeric_scale
+              FROM information_schema.COLUMNS c LEFT JOIN information_schema.element_types e
+                 ON ((c.table_catalog, c.table_schema, c.table_name, 'TABLE', c.dtd_identifier)
+                     = (e.object_catalog, e.object_schema, e.object_name, e.object_type, e.collection_type_identifier))
+              where table_schema = $1 and table_name = $2 ORDER BY c.ordinal_position;`
+	return db.Query(q, table.schema, table.name)
+}
+
+func processColumns(conv *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
+	var colDefault, elementDataType sql.NullString
+	var charMaxLen, numericPrecision, numericScale sql.NullInt64
+	for cols.Next() {
+		err := cols.Scan(&colName, &dataType, &elementDataType, &isNullable, &colDefault, &charMaxLen, &numericPrecision, &numericScale)
+		if err != nil {
+			fmt.Printf("Can't scan: %v\n", err)
+			continue
+		}
+		unique := false
+		ignored := schema.Ignored{}
+		for _, c := range constraints[colName] {
+			// c can be UNIQUE, PRIMARY KEY, FOREIGN KEY,
+			// or CHECK (based on msql, sql server, postgres docs).
+			// We've already filtered out PRIMARY KEY.
+			switch c {
+			case "UNIQUE":
+				unique = true
+			case "FOREIGN KEY":
+				ignored.ForeignKey = true
+			case "CHECK":
+				ignored.Check = true
+			}
+		}
+		ignored.Default = colDefault.Valid
+		c := schema.Column{
+			Name:    colName,
+			Type:    toType(dataType, elementDataType, charMaxLen, numericPrecision, numericScale),
+			NotNull: toNotNull(conv, isNullable),
+			Unique:  unique,
+			Ignored: ignored,
+		}
+		colDefs[colName] = c
+		colNames = append(colNames, colName)
+	}
+	return colDefs, colNames
+}
+
+// getConstraints returns a list of primary keys and by-column map of
+// other constraints.  Note: we need to preserve ordinal order of
+// columns in primary key constraints.
+func getConstraints(conv *Conv, db *sql.DB, table schemaAndName) ([]string, map[string][]string, error) {
+	q := `SELECT k.COLUMN_NAME, t.CONSTRAINT_TYPE
+              FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t
+                INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS k
+                  ON t.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND t.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
+              WHERE k.TABLE_SCHEMA = $1 AND k.TABLE_NAME = $2 ORDER BY k.ordinal_position;`
+	rows, err := db.Query(q, table.schema, table.name)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer rows.Close()
+	var primaryKeys []string
+	var col, constraint string
+	m := make(map[string][]string)
+	for rows.Next() {
+		err := rows.Scan(&col, &constraint)
+		if err != nil {
+			fmt.Printf("Can't scan: %v\n", err)
+			continue
+		}
+		if col == "" || constraint == "" {
+			fmt.Printf("Got empty col or constraint\n")
+			continue
+		}
+		switch constraint {
+		case "PRIMARY KEY":
+			primaryKeys = append(primaryKeys, col)
+		default:
+			m[col] = append(m[col], constraint)
+		}
+	}
+	return primaryKeys, m, nil
+}
+
+func toType(dataType string, elementDataType sql.NullString, charLen sql.NullInt64, numericPrecision, numericScale sql.NullInt64) schema.Type {
+	switch {
+	case dataType == "ARRAY" && elementDataType.Valid:
+		return schema.Type{Name: elementDataType.String, ArrayBounds: []int64{-1}}
+		// TODO: handle error cases.
+		// TODO: handle case of multiple array bounds.
+	case charLen.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{charLen.Int64}}
+	case dataType == "numeric" && numericPrecision.Valid && numericScale.Valid && numericScale.Int64 != 0:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64, numericScale.Int64}}
+	case dataType == "numeric" && numericPrecision.Valid:
+		return schema.Type{Name: dataType, Mods: []int64{numericPrecision.Int64}}
+	default:
+		return schema.Type{Name: dataType}
+	}
+}
+
+func toNotNull(conv *Conv, isNullable string) bool {
+	switch isNullable {
+	case "YES":
+		return false
+	case "NO":
+		return true
+	}
+	conv.unexpected(fmt.Sprintf("isNullable column has unknown value: %s", isNullable))
+	return false
+}
diff --git a/internal/report.go b/internal/report.go
index 6da0398..3a44ea5 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -26,7 +26,7 @@ import (
 
 // GenerateReport analyzes schema and data conversion stats and writes a
 // detailed report to w and returns a brief summary (as a string).
-func GenerateReport(conv *Conv, w *bufio.Writer, badWrites map[string]int64) string {
+func GenerateReport(fromPgDump bool, conv *Conv, w *bufio.Writer, badWrites map[string]int64) string {
 	reports := analyzeTables(conv, badWrites)
 	summary := generateSummary(conv, reports, badWrites)
 	writeHeading(w, "Summary of Conversion")
@@ -39,14 +39,19 @@ func GenerateReport(conv *Conv, w *bufio.Writer, badWrites map[string]int64) str
 			strings.Join(ignored, ", ")), 80, 0)
 		w.WriteString("\n\n")
 	}
-	justifyLines(w, "The remainder of this report provides stats on "+
-		"the pg_dump statements processed, followed by a table-by-table "+
-		"listing of schema and data conversion details. "+
+	statementsMsg := ""
+	if fromPgDump {
+		statementsMsg = "stats on the pg_dump statements processed, followed by "
+	}
+	justifyLines(w, "The remainder of this report provides "+statementsMsg+
+		"a table-by-table listing of schema and data conversion details. "+
 		"For background on the schema and data conversion process used, "+
 		"and explanations of the terms and notes used in this "+
 		"report, see HarbourBridge's README.", 80, 0)
 	w.WriteString("\n\n")
-	writeStmtStats(conv, w)
+	if fromPgDump {
+		writeStmtStats(conv, w)
+	}
 	for _, t := range reports {
 		h := fmt.Sprintf("Table %s", t.srcTable)
 		if t.srcTable != t.spTable {
diff --git a/internal/toddl.go b/internal/toddl.go
index 7bfffd7..47a7cd7 100644
--- a/internal/toddl.go
+++ b/internal/toddl.go
@@ -91,13 +91,13 @@ func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []schem
 		}
 	}
 	switch id {
-	case "bool":
+	case "bool", "boolean":
 		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).
+	case "bpchar", "character": // 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
@@ -110,19 +110,19 @@ func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []schem
 	case "date":
 		maxExpectedMods(0)
 		return ddl.Date{}, nil
-	case "float8":
+	case "float8", "double precision":
 		maxExpectedMods(0)
 		return ddl.Float64{}, nil
-	case "float4":
+	case "float4", "real":
 		maxExpectedMods(0)
 		return ddl.Float64{}, []schemaIssue{widened}
-	case "int8":
+	case "int8", "bigint":
 		maxExpectedMods(0)
 		return ddl.Int64{}, nil
-	case "int4":
+	case "int4", "integer":
 		maxExpectedMods(0)
 		return ddl.Int64{}, []schemaIssue{widened}
-	case "int2":
+	case "int2", "smallint":
 		maxExpectedMods(0)
 		return ddl.Int64{}, []schemaIssue{widened}
 	case "numeric": // Map all numeric types to float64.
@@ -140,14 +140,14 @@ func toSpannerType(conv *Conv, id string, mods []int64) (ddl.ScalarType, []schem
 	case "text":
 		maxExpectedMods(0)
 		return ddl.String{Len: ddl.MaxLength{}}, nil
-	case "timestamptz":
+	case "timestamptz", "timestamp with time zone":
 		maxExpectedMods(1)
 		return ddl.Timestamp{}, nil
-	case "timestamp":
+	case "timestamp", "timestamp without time zone":
 		maxExpectedMods(1)
 		// Map timestamp without timezone to Spanner timestamp.
 		return ddl.Timestamp{}, []schemaIssue{timestamp}
-	case "varchar":
+	case "varchar", "character varying":
 		maxExpectedMods(1)
 		if len(mods) > 0 {
 			return ddl.String{Len: ddl.Int64Length{Value: mods[0]}}, nil
diff --git a/main.go b/main.go
index 8989606..426639c 100644
--- a/main.go
+++ b/main.go
@@ -52,13 +52,16 @@ var (
 	dbNameOverride   string
 	instanceOverride string
 	filePrefix       = ""
+	driverName       = ""
 	verbose          bool
+	fromPgDump       bool
 )
 
 func init() {
-	flag.StringVar(&dbNameOverride, "dbname", "", "dbname: name to use for Spanner db")
+	flag.StringVar(&dbNameOverride, "dbname", "", "dbname: name to use for Spanner DB")
 	flag.StringVar(&instanceOverride, "instance", "", "instance: Spanner instance to use")
 	flag.StringVar(&filePrefix, "prefix", "", "prefix: file prefix for generated files")
+	flag.StringVar(&driverName, "driver", "", "driver name: experimental flag for accessing source DB via database/sql driver (only accepted value is \"postgres\")")
 	flag.BoolVar(&verbose, "v", false, "verbose: print additional output")
 }
 
@@ -116,19 +119,58 @@ func main() {
 		filePrefix = dbName + "."
 	}
 
-	err = process(project, instance, dbName, nil, filePrefix, now)
+	// If driverName specified, access source DB via database/sql
+	// driver. Otherwise read pgdump data from stdin.
+	fromPgDump = driverName == ""
+	if fromPgDump {
+		err = pgDumpToSpanner(project, instance, dbName, nil, filePrefix, now)
+	} else {
+		err = sourceToSpanner(driverName, project, instance, dbName, filePrefix, now)
+	}
+
 	if err != nil {
 		panic(err)
 	}
 }
 
+// sourceToSpanner migrates a source DB (accessed via a database/sql
+// driver) to Spanner.  The source DB must support the
+// information_schema standard (for information about tables, views,
+// columns etc). Data is accessed via SELECT statements.
+// TODO: add data conversion.
+func sourceToSpanner(driver, projectID, instanceID, dbName string, outputFilePrefix string, now time.Time) error {
+	fmt.Println(`###################################################################
+Accessing a source DB via an database/sql driver is an experimental
+feature that is not fully implemented. Currently we only do schema
+conversion and we only support postgres.
+###################################################################`)
+	if driver != "postgres" {
+		return fmt.Errorf("Driver %s not supported", driver)
+	}
+	conv := internal.MakeConv()
+	err := internal.ProcessInfoSchema(conv, driver)
+	if err != nil {
+		return err
+	}
+	writeSchemaFile(conv, now, outputFilePrefix+schemaFile)
+	db, err := createDatabase(projectID, instanceID, dbName, conv)
+	if err != nil {
+		fmt.Printf("\nCan't create database: %v\n", err)
+		return fmt.Errorf("can't create database")
+	}
+	badWrites := make(map[string]int64) // Empty bad writes since no data conversion yet.
+	banner := getBanner(now, db)
+	report(badWrites, 0, banner, conv, outputFilePrefix+reportFile)
+	return nil
+}
+
 type ioStreams struct {
 	in, out *os.File
 }
 
 var ioHelper = &ioStreams{os.Stdin, os.Stdout}
 
-func process(projectID, instanceID, dbName string, helper *ioStreams, ouputFilePrefix string, now time.Time) error {
+func pgDumpToSpanner(projectID, instanceID, dbName string, helper *ioStreams, outputFilePrefix string, now time.Time) error {
 	if helper != nil {
 		ioHelper = helper
 	}
@@ -146,7 +188,7 @@ func process(projectID, instanceID, dbName string, helper *ioStreams, ouputFileP
 		fmt.Fprintf(helper.out, "Failed to parse the data file: %v", err)
 		return fmt.Errorf("failed to parse the data file")
 	}
-	writeSchemaFile(conv, now, ouputFilePrefix+schemaFile)
+	writeSchemaFile(conv, now, outputFilePrefix+schemaFile)
 	db, err := createDatabase(projectID, instanceID, dbName, conv)
 	if err != nil {
 		fmt.Printf("\nCan't create database: %v\n", err)
@@ -164,13 +206,15 @@ func process(projectID, instanceID, dbName string, helper *ioStreams, ouputFileP
 	}
 	rows := conv.Rows()
 	bw := secondPass(f, client, conv, rows)
+	banner := getBanner(now, db)
 	// TODO(hengfeng): When we refactor `process` into a separate module, and
 	// the parameters will capture everything we need from main.
-	report(bw, n, now, db, conv, ouputFilePrefix+reportFile, ouputFilePrefix+badDataFile)
+	report(bw.DroppedRowsByTable(), n, banner, conv, outputFilePrefix+reportFile)
+	writeBadData(bw, conv, banner, outputFilePrefix+badDataFile)
 	return nil
 }
 
-func report(bw *spanner.BatchWriter, bytesRead int64, now time.Time, db string, conv *internal.Conv, reportFileName string, badDataFileName string) {
+func report(badWrites map[string]int64, bytesRead int64, banner string, conv *internal.Conv, reportFileName string) {
 	f, err := os.Create(reportFileName)
 	if err != nil {
 		fmt.Fprintf(ioHelper.out, "Can't write out report file %s: %v\n", reportFileName, err)
@@ -180,20 +224,22 @@ func report(bw *spanner.BatchWriter, bytesRead int64, now time.Time, db string,
 		defer f.Close()
 	}
 	w := bufio.NewWriter(f)
-	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 := internal.GenerateReport(conv, w, badWrites)
+	summary := internal.GenerateReport(fromPgDump, 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())
+	if fromPgDump {
+		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())
+	} else {
+		fmt.Fprintf(ioHelper.out, "Processed source database via %s driver (%d rows of data, %d unexpected conditions).\n",
+			driverName, conv.Rows(), conv.Unexpecteds())
+	}
 	// We've already written summary to f (as part of GenerateReport).
 	// In the case where f is stdout, don't write a duplicate copy.
 	if f != ioHelper.out {
 		fmt.Fprint(ioHelper.out, summary)
 		fmt.Fprintf(ioHelper.out, "See file '%s' for details of the schema and data conversions.\n", reportFileName)
 	}
-	writeBadData(bw, conv, banner, badDataFileName)
 }
 
 func firstPass(f *os.File, fileSize int64, conv *internal.Conv) error {
@@ -560,3 +606,7 @@ func sum(m map[string]int64) int64 {
 	}
 	return n
 }
+
+func getBanner(now time.Time, db string) string {
+	return fmt.Sprintf("Generated at %s for db %s\n\n", now.Format("2006-01-02 15:04:05"), db)
+}


Test patch:
diff --git a/integration_test.go b/integration_test.go
index d2e197d..1ae99f0 100644
--- a/integration_test.go
+++ b/integration_test.go
@@ -131,7 +131,7 @@ func TestIntegration_SimpleUse(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to open the test data file: %v", err)
 	}
-	err = process(projectID, instanceID, dbName, &ioStreams{f, os.Stdout}, filePrefix, now)
+	err = pgDumpToSpanner(projectID, instanceID, dbName, &ioStreams{f, os.Stdout}, filePrefix, now)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/internal/report_test.go b/internal/report_test.go
index 9966fb9..e88d074 100644
--- a/internal/report_test.go
+++ b/internal/report_test.go
@@ -54,7 +54,7 @@ func TestReport(t *testing.T) {
 	conv.stats.unexpected["Testing unexpected messages"] = 5
 	buf := new(bytes.Buffer)
 	w := bufio.NewWriter(buf)
-	GenerateReport(conv, w, badWrites)
+	GenerateReport(true, conv, w, badWrites)
 	w.Flush()
 	// Print copy of report to stdout (shows up when running go test -v).
 	fmt.Print(buf.String())