Repository: cloudspannerecosystem/harbourbridge
Pull request #521 (2023-04-26 13:25:43)

Structured Reports for Harbourbridge
* Adds structured reports for Harbourbridge. The reports generated via harbourbridge have been refactored into a `internal/reports` package with this change.

* The new `StructuredReport` object is used as a source of truth to generate the text report. The text report is nearly identical to the existing text report generated by harbourbridge ,with minor changes (discussed as part of design).

* Unit test have been modified to load and assert reports from file for maintainability.



Patch:
diff --git a/README.md b/README.md
index c070e55..0659d51 100644
--- a/README.md
+++ b/README.md
@@ -54,7 +54,7 @@ anyone who can access the created database.**
 
 As it processes the data, HarbourBridge reports on progress, provides
 stats on the schema and data conversion steps, and an overall assessment of the
-quality of the conversion. It also generates a schema file, report file and
+quality of the conversion. It also generates a schema file, report files and
 a session file (and a bad-data file if data was dropped). See
 [Files Generated by HarbourBridge](#files-generated-by-harbourbridge). Details
 of how source database's schema is mapped to Spanner can be found in the
@@ -207,7 +207,7 @@ See the [Troubleshooting Guide](#troubleshooting-guide) for help on debugging
 issues.
 
 HarbourBridge also [generates several files](#files-generated-by-harbourbridge)
-when it runs: a schema file, a report file (with detailed analysis of the
+when it runs: a schema file, report files (with detailed analysis of the
 conversion), a session file and a bad data file (if any data was dropped).
 
 ### Setting up the emulator
@@ -297,15 +297,20 @@ HarbourBridge generates several files as it runs:
 
 - Schema file (ending in `schema.txt`): contains the generated Spanner
   schema, interspersed with comments that cross-reference to the relevant
-  PostgreSQL/MySQL schema definitions.
+  source schema definitions.
 
 - Session file (ending in `session.json`): contains all schema and data
   conversion state endcoded as JSON. It is basically a snapshot of the session.
 
-- Report file (ending in `report.txt`): contains a detailed analysis of the
-  PostgreSQL/MySQL to Spanner migration, including table-by-table stats and an
-  analysis of PostgreSQL/MySQL types that don't cleanly map onto Spanner types.
-  Note that PostgreSQL/MySQL types that don't have a corresponding Spanner type
+- Structured Report file (ending in `structured_report.json`): contains a JSON based
+structured analysis of the source to Spanner migration. The structured report can be
+used to in-depth analysis of Harbourbridge findings via BI tools. For a detailed
+description of each element of a report, refer to [Elements of a Report](/internal/reports/REPORT.md).
+
+- Text Report file (ending in `report.txt`): contains a detailed analysis of the
+  source to Spanner migration, including table-by-table stats and an
+  analysis of Source types that don't cleanly map onto Spanner types.
+  Note that source types that don't have a corresponding Spanner type
   are mapped to STRING(MAX).
 
 - Bad data file (ending in `dropped.txt`): contains details of data
diff --git a/cmd/data.go b/cmd/data.go
index da1cb61..e5d9cb7 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -155,9 +155,9 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 
 	// If filePrefix not explicitly set, use dbName as prefix.
 	if cmd.filePrefix == "" {
-		cmd.filePrefix = targetProfile.Conn.Sp.Dbname + "."
+		cmd.filePrefix = targetProfile.Conn.Sp.Dbname
 	}
-	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
+	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix, dbName, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
 	// Cleanup hb tmp data directory.
 	os.RemoveAll(filepath.Join(os.TempDir(), constants.HB_TMP_DIR))
diff --git a/cmd/schema.go b/cmd/schema.go
index e27e7a0..feedd40 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -101,7 +101,7 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 
 	// If filePrefix not explicitly set, use generated dbName.
 	if cmd.filePrefix == "" {
-		cmd.filePrefix = dbName + "."
+		cmd.filePrefix = dbName
 	}
 
 	schemaConversionStartTime := time.Now()
@@ -129,7 +129,7 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 	schemaCoversionEndTime := time.Now()
 	conv.Audit.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
 	banner := utils.GetBanner(schemaConversionStartTime, dbName)
-	conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
+	conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, banner, conv, cmd.filePrefix, dbName, ioHelper.Out)
 	// Cleanup hb tmp data directory.
 	os.RemoveAll(filepath.Join(os.TempDir(), constants.HB_TMP_DIR))
 	return subcommands.ExitSuccess
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index 9136a7f..fda806f 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -107,7 +107,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 
 	// If filePrefix not explicitly set, use dbName as prefix.
 	if cmd.filePrefix == "" {
-		cmd.filePrefix = dbName + "."
+		cmd.filePrefix = dbName
 	}
 
 	var (
@@ -132,7 +132,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	conv.Audit.SkipMetricsPopulation = os.Getenv("SKIP_METRICS_POPULATION") == "true"
 
 	if !cmd.dryRun {
-		conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix+reportFile, ioHelper.Out)
+		conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix, dbName, ioHelper.Out)
 		bw, err = MigrateDatabase(ctx, targetProfile, sourceProfile, dbName, &ioHelper, cmd, conv, nil)
 		if err != nil {
 			err = fmt.Errorf("can't finish database migration for db %s: %v", dbName, err)
@@ -155,7 +155,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		conv.Audit.DataConversionDuration = dataCoversionEndTime.Sub(schemaCoversionEndTime)
 		banner = utils.GetBanner(schemaConversionStartTime, dbName)
 	}
-	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
+	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix, dbName, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
 
 	// Cleanup hb tmp data directory.
diff --git a/cmd/utils.go b/cmd/utils.go
index 32beeb8..44aa820 100644
--- a/cmd/utils.go
+++ b/cmd/utils.go
@@ -29,10 +29,9 @@ import (
 )
 
 var (
-	badDataFile = "dropped.txt"
-	reportFile  = "report.txt"
-	schemaFile  = "schema.txt"
-	sessionFile = "session.json"
+	badDataFile = ".dropped.txt"
+	schemaFile  = ".schema.txt"
+	sessionFile = ".session.json"
 )
 
 const (
diff --git a/common/metrics/metrics.go b/common/metrics/metrics.go
index 373bc80..cbb33c4 100644
--- a/common/metrics/metrics.go
+++ b/common/metrics/metrics.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/internal/reports"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 )
 
@@ -46,7 +47,7 @@ func getMigrationDataSchemaPatterns(conv *internal.Conv, migrationData *migratio
 		if len(srcSchema.PrimaryKeys) == 0 {
 			numMissingPrimaryKey++
 		}
-		_, cols, warnings := internal.AnalyzeCols(conv, srcTableId)
+		_, cols, warnings := reports.AnalyzeCols(conv, srcTableId)
 		numColumns += int32(cols)
 		numWarnings += int32(warnings)
 	}
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 26ba029..f75502b 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -49,6 +49,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/common/metrics"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/internal/reports"
 	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -373,8 +374,25 @@ func populateDataConv(conv *internal.Conv, config writer.BatchWriterConfig, clie
 }
 
 // Report generates a report of schema and data conversion.
-func Report(driver string, badWrites map[string]int64, BytesRead int64, banner string, conv *internal.Conv, reportFileName string, out *os.File) {
-	f, err := os.Create(reportFileName)
+func Report(driver string, badWrites map[string]int64, BytesRead int64, banner string, conv *internal.Conv, reportFileName string, dbName string, out *os.File) {
+
+	//Write the structured report file
+	structuredReportFileName := fmt.Sprintf("%s.%s", reportFileName, "structured_report.json")
+	structuredReport := reports.GenerateStructuredReport(driver, dbName, conv, badWrites, true, true)
+	fBytes, _ := json.MarshalIndent(structuredReport, "", " ")
+	f, err := os.Create(structuredReportFileName)
+	if err != nil {
+		fmt.Fprintf(out, "Can't write out structured report file %s: %v\n", reportFileName, err)
+		fmt.Fprintf(out, "Writing report to stdout\n")
+		f = out
+	} else {
+		defer f.Close()
+	}
+	f.Write(fBytes)
+
+	//Write the text report file from the structured report
+	textReportFileName := fmt.Sprintf("%s.%s", reportFileName, "report.txt")
+	f, err = os.Create(textReportFileName)
 	if err != nil {
 		fmt.Fprintf(out, "Can't write out report file %s: %v\n", reportFileName, err)
 		fmt.Fprintf(out, "Writing report to stdout\n")
@@ -384,9 +402,9 @@ func Report(driver string, badWrites map[string]int64, BytesRead int64, banner s
 	}
 	w := bufio.NewWriter(f)
 	w.WriteString(banner)
-
-	summary := internal.GenerateReport(driver, conv, w, badWrites, true, true)
+	reports.GenerateTextReport(structuredReport, w)
 	w.Flush()
+
 	var isDump bool
 	if strings.Contains(driver, "dump") {
 		isDump = true
@@ -401,7 +419,7 @@ func Report(driver string, badWrites map[string]int64, BytesRead int64, banner s
 	// 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 != out {
-		fmt.Fprint(out, summary)
+		fmt.Fprint(out, structuredReport.Summary.Text)
 		fmt.Fprintf(out, "See file '%s' for details of the schema and data conversions.\n", reportFileName)
 	}
 }
@@ -767,8 +785,8 @@ func WriteConvGeneratedFiles(conv *internal.Conv, dbName string, driver string,
 	}
 	schemaFileName := dirPath + dbName + "_schema.txt"
 	WriteSchemaFile(conv, now, schemaFileName, out)
-	reportFileName := dirPath + dbName + "_report.txt"
-	Report(driver, nil, BytesRead, "", conv, reportFileName, out)
+	reportFileName := dirPath + dbName
+	Report(driver, nil, BytesRead, "", conv, reportFileName, dbName, out)
 	sessionFileName := dirPath + dbName + ".session.json"
 	WriteSessionFile(conv, sessionFileName, out)
 	return dirPath, nil
diff --git a/internal/helpers.go b/internal/helpers.go
index 869a876..e916c47 100644
--- a/internal/helpers.go
+++ b/internal/helpers.go
@@ -28,8 +28,8 @@ type Counter struct {
 
 var Cntr Counter
 
-// contains check string present in list.
-func contains(l []string, str string) bool {
+// Contains check string present in list.
+func Contains(l []string, str string) bool {
 	for _, s := range l {
 		if s == str {
 			return true
diff --git a/internal/report.go b/internal/report.go
deleted file mode 100644
index 38cea59..0000000
--- a/internal/report.go
+++ /dev/null
@@ -1,937 +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 internal
-
-import (
-	"bufio"
-	"fmt"
-	"sort"
-	"strconv"
-	"strings"
-
-	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
-	"github.com/cloudspannerecosystem/harbourbridge/schema"
-	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
-)
-
-// GenerateReport analyzes schema and data conversion stats and writes a
-// detailed report to w and returns a brief summary (as a string).
-func GenerateReport(driverName string, conv *Conv, w *bufio.Writer, badWrites map[string]int64, printTableReports bool, printUnexpecteds bool) string {
-	reports := AnalyzeTables(conv, badWrites)
-	summary := GenerateSummary(conv, reports, badWrites)
-	writeHeading(w, "Summary of Conversion")
-	w.WriteString(summary)
-	ignored := IgnoredStatements(conv)
-	w.WriteString("\n")
-	w.WriteString(conversionDuration(conv, w))
-	if len(ignored) > 0 {
-		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")
-	}
-	migrationType := ""
-	if *conv.Audit.MigrationType == migration.MigrationData_DATA_ONLY {
-		migrationType = "data"
-	} else if *conv.Audit.MigrationType == migration.MigrationData_SCHEMA_ONLY {
-		migrationType = "schema"
-	} else if *conv.Audit.MigrationType == migration.MigrationData_SCHEMA_AND_DATA {
-		migrationType = "schema and data"
-	}
-	statementsMsg := ""
-	var isDump bool
-	if strings.Contains(driverName, "dump") {
-		isDump = true
-	}
-	if isDump {
-		statementsMsg = "stats on the " + driverName + " statements processed, followed by "
-	}
-	justifyLines(w, "The remainder of this report provides "+statementsMsg+
-		"a table-by-table listing of "+migrationType+" conversion details. "+
-		"For background on the "+migrationType+" conversion process used, "+
-		"and explanations of the terms and notes used in this "+
-		"report, see HarbourBridge's README.", 80, 0)
-	w.WriteString("\n\n")
-	if isDump {
-		writeStmtStats(driverName, conv, w)
-	}
-	reportNameChanges(conv, w)
-
-	if printTableReports {
-		for _, t := range reports {
-			h := fmt.Sprintf("Table %s", conv.SpSchema[t.SrcTable].Name)
-			if t.SrcTable != t.SpTable {
-				h = h + fmt.Sprintf(" (mapped to Spanner table %s)", t.SpTable)
-			}
-			writeHeading(w, h)
-			w.WriteString(rateConversion(t.rows, t.badRows, t.Cols, t.Warnings, t.SyntheticPKey != "", false, conv.SchemaMode(), *conv.Audit.MigrationType, conv.Audit.DryRun))
-			w.WriteString("\n")
-			for _, x := range t.Body {
-				fmt.Fprintf(w, "%s\n", x.Heading)
-				for i, l := range x.Lines {
-					justifyLines(w, fmt.Sprintf("%d) %s.\n", i+1, l), 80, 3)
-				}
-				w.WriteString("\n")
-			}
-		}
-	}
-
-	if printUnexpecteds {
-		writeUnexpectedConditions(driverName, conv, w)
-	}
-
-	if conv.Audit.StreamingStats.Streaming {
-		writeStreamingMigrationReport(driverName, conv, w)
-	}
-	return summary
-}
-
-func sumNestedMapValues(mp map[string]map[string]int64) int64 {
-	n := int64(0)
-	for _, x := range mp {
-		for _, y := range x {
-			n += y
-		}
-	}
-	return n
-}
-
-func getStreamName(driver string) (string, error) {
-	switch driver {
-	case constants.DYNAMODB:
-		return "DynamoDB Streams", nil
-	default:
-		return "", fmt.Errorf("streaming migration report only printed for DynamoDB currently")
-	}
-}
-
-func getRecordTypes(driver string) []string {
-	switch driver {
-	case constants.DYNAMODB:
-		return []string{"INSERT", "REMOVE", "MODIFY"}
-	default:
-		return []string{}
-	}
-}
-
-// writeStreamingMigrationReport writes summary of streaming migration process to report file
-// generated by HarbourBridge.
-func writeStreamingMigrationReport(driverName string, conv *Conv, w *bufio.Writer) {
-	streamName, err := getStreamName(driverName)
-	if err != nil {
-		return
-	}
-
-	w.WriteString("----------------------------------\n")
-	w.WriteString("Summary of streaming migration\n")
-	w.WriteString("----------------------------------\n")
-
-	stats := conv.Audit.StreamingStats
-
-	tableIds := ddl.GetSortedTableIdsBySpName(conv.SpSchema)
-
-	w.WriteString(fmt.Sprintf("Processed %d out of %d %s successfully.\n", len(stats.TotalRecords), len(tableIds), streamName))
-
-	totalReadRecords := sumNestedMapValues(stats.TotalRecords)
-	w.WriteString(fmt.Sprintf("\nCount of records read from %s: %s\n", streamName, strconv.FormatInt(totalReadRecords, 10)))
-
-	totalWrittenRecords := totalReadRecords - sumNestedMapValues(stats.BadRecords) - sumNestedMapValues(stats.DroppedRecords)
-	w.WriteString(fmt.Sprintf("Count of records written to Cloud Spanner successfully: %s\n", strconv.FormatInt(totalWrittenRecords, 10)))
-
-	recordTypes := getRecordTypes(driverName)
-
-	w.WriteString(fmt.Sprintf("\nTablewise summary of processing of %s (Written records / Total records)\nbroken down by record type.\n\n", streamName))
-
-	colWidth := make(map[string]int)
-
-	colWidth["Table"] = len("Table")
-	for srcTable := range stats.TotalRecords {
-		if len(srcTable) > colWidth["Table"] {
-			colWidth["Table"] = len(srcTable)
-		}
-	}
-	// Add padding of 2 spaces to left and 2 to right for a column value.
-	colWidth["Table"] += 4
-
-	for _, change := range recordTypes {
-		colWidth[change] = len(change)
-		for srcTable, records := range stats.TotalRecords {
-			total := records[change]
-			success := total - stats.BadRecords[srcTable][change] - stats.DroppedRecords[srcTable][change]
-
-			progress := fmt.Sprintf("%s/%s", strconv.FormatInt(success, 10), strconv.FormatInt(total, 10))
-			if len(progress) > colWidth[change] {
-				colWidth[change] = len(progress)
-			}
-		}
-		colWidth[change] += 4
-	}
-
-	totalWidth := len(colWidth) + 1
-	for _, width := range colWidth {
-		totalWidth += width
-	}
-
-	seperator := strings.Repeat("-", totalWidth) + "\n"
-
-	recordTypes = append([]string{"Table"}, recordTypes...)
-
-	w.WriteString(seperator)
-	for _, colName := range recordTypes {
-		spaces := colWidth[colName] - len(colName)
-		leftSpace := spaces / 2
-		rightSpace := spaces - leftSpace
-
-		w.WriteString(fmt.Sprintf("|%s%s%s", strings.Repeat(" ", leftSpace), colName, strings.Repeat(" ", rightSpace)))
-	}
-	w.WriteString("|\n" + seperator)
-
-	for srcTable, records := range stats.TotalRecords {
-		for _, colName := range recordTypes {
-			var progress string
-			if colName == "Table" {
-				progress = srcTable
-			} else {
-				total := records[colName]
-				success := total - stats.BadRecords[srcTable][colName] - stats.DroppedRecords[srcTable][colName]
-				progress = fmt.Sprintf("%s/%s", strconv.FormatInt(success, 10), strconv.FormatInt(total, 10))
-			}
-
-			spaces := colWidth[colName] - len(progress)
-			leftSpace := spaces / 2
-			rightSpace := (spaces - leftSpace)
-
-			w.WriteString(fmt.Sprintf("|%s%s%s", strings.Repeat(" ", leftSpace), progress, strings.Repeat(" ", rightSpace)))
-		}
-		w.WriteString("|\n" + seperator)
-	}
-}
-
-type tableReport struct {
-	SrcTable      string
-	SpTable       string
-	rows          int64
-	badRows       int64
-	Cols          int64
-	Warnings      int64
-	SyntheticPKey string // Empty string means no synthetic primary key was needed.
-	Body          []tableReportBody
-}
-
-type tableReportBody struct {
-	Heading string
-	Lines   []string
-}
-
-// AnalyzeTables generates table reports for all processed tables.
-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 tableNames []string
-	for _, srcTable := range conv.SrcSchema {
-		tableNames = append(tableNames, srcTable.Name)
-	}
-	sort.Strings(tableNames)
-	for _, tableName := range tableNames {
-		tableId, err := GetTableIdFromSrcName(conv.SrcSchema, tableName)
-		if err != nil {
-			continue
-		}
-		if _, isPresent := conv.SpSchema[tableId]; isPresent {
-			r = append(r, buildTableReport(conv, tableId, badWrites))
-		}
-	}
-	return r
-}
-
-func buildTableReport(conv *Conv, tableId string, badWrites map[string]int64) tableReport {
-	srcSchema, ok1 := conv.SrcSchema[tableId]
-	spSchema, ok2 := conv.SpSchema[tableId]
-	tr := tableReport{SrcTable: tableId, SpTable: tableId}
-	if !ok1 || !ok2 {
-		m := "bad source-DB-to-Spanner table mapping or Spanner schema"
-		conv.Unexpected("report: " + m)
-		tr.Body = []tableReportBody{{Heading: "Internal error: " + m}}
-		return tr
-	}
-	if *conv.Audit.MigrationType != migration.MigrationData_DATA_ONLY {
-		issues, cols, warnings := AnalyzeCols(conv, tableId)
-		tr.Cols = cols
-		tr.Warnings = warnings
-		if pk, ok := conv.SyntheticPKeys[tableId]; ok {
-			tr.SyntheticPKey = pk.ColId
-			synthColName := conv.SpSchema[tableId].ColDefs[pk.ColId].Name
-			tr.Body = buildTableReportBody(conv, tableId, issues, spSchema, srcSchema, &synthColName, nil)
-		} else if pk, ok := conv.UniquePKey[tableId]; ok {
-			tr.Body = buildTableReportBody(conv, tableId, issues, spSchema, srcSchema, nil, pk)
-		} else {
-			tr.Body = buildTableReportBody(conv, tableId, issues, spSchema, srcSchema, nil, nil)
-		}
-
-	}
-	if !conv.SchemaMode() {
-		fillRowStats(conv, tableId, badWrites, &tr)
-	}
-	return tr
-}
-
-func buildTableReportBody(conv *Conv, tableId string, issues map[string][]SchemaIssue, spSchema ddl.CreateTable, srcSchema schema.Table, syntheticPK *string, uniquePK []string) []tableReportBody {
-	var body []tableReportBody
-	for _, p := range []struct {
-		heading  string
-		severity severity
-	}{
-		{"Warning", warning},
-		{"Note", note},
-		{"Suggestion", suggestion},
-		{"Error", errors},
-	} {
-		// Print out issues is alphabetical column order.
-		var colNames []string
-		for colId := range issues {
-			colNames = append(colNames, conv.SpSchema[tableId].ColDefs[colId].Name)
-		}
-		sort.Strings(colNames)
-		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 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))
-			}
-		}
-		if uniquePK != nil {
-			// Warning about using a column with unique constraint as primary key
-			// in case primary key is absent.
-			if p.severity == warning {
-				l = append(l, fmt.Sprintf("UNIQUE constraint on column(s) '%s' replaced with primary key since this table didn't have one. Spanner requires a primary key for every table", strings.Join(uniquePK, ", ")))
-			}
-		}
-
-		if p.severity == warning {
-			for _, spFk := range conv.SpSchema[tableId].ForeignKeys {
-				srcFk, err := GetSrcFkFromId(conv.SrcSchema[tableId].ForeignKeys, spFk.Id)
-				if err != nil {
-					continue
-				}
-				_, isChanged := FixName(srcFk.Name)
-				if isChanged && srcFk.Name != spFk.Name {
-					l = append(l, fmt.Sprintf("%s, Foreign Key '%s' is mapped to '%s'", IssueDB[IllegalName].Brief, srcFk.Name, spFk.Name))
-				}
-			}
-			for _, spIdx := range conv.SpSchema[tableId].Indexes {
-				srcIdx, err := GetSrcIndexFromId(conv.SrcSchema[tableId].Indexes, spIdx.Id)
-				if err != nil {
-					continue
-				}
-				_, isChanged := FixName(srcIdx.Name)
-				if isChanged && srcIdx.Name != spIdx.Name {
-					l = append(l, fmt.Sprintf("%s, Index '%s' is mapped to '%s'", IssueDB[IllegalName].Brief, srcIdx.Name, spIdx.Name))
-				}
-			}
-
-			_, isChanged := FixName(srcSchema.Name)
-			if isChanged && (spSchema.Name != srcSchema.Name) {
-				l = append(l, fmt.Sprintf("%s, Table '%s' is mapped to '%s'", IssueDB[IllegalName].Brief, srcSchema.Name, spSchema.Name))
-			}
-		}
-
-		issueBatcher := make(map[SchemaIssue]bool)
-		for _, colName := range colNames {
-			colId, _ := GetColIdFromSpName(conv.SpSchema[tableId].ColDefs, colName)
-			for _, i := range issues[colId] {
-				if IssueDB[i].severity != p.severity {
-					continue
-				}
-				if IssueDB[i].batch {
-					if issueBatcher[i] {
-						// Have already reported a previous instance of this
-						// (batched) issue, so skip this one.
-						continue
-					}
-					issueBatcher[i] = true
-				}
-				srcColType := srcSchema.ColDefs[colId].Type.Print()
-				spColType := spSchema.ColDefs[colId].T.PrintColumnDefType()
-				if conv.SpDialect == constants.DIALECT_POSTGRESQL {
-					spColType = spSchema.ColDefs[colId].T.PGPrintColumnDefType()
-				}
-				srcColName := srcSchema.ColDefs[colId].Name
-				spColName := spSchema.ColDefs[colId].Name
-
-				// 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.
-				// TODO: add logic to choose case for Spanner types based
-				// on case of srcType.
-				spColType = strings.ToLower(spColType)
-				switch i {
-				case DefaultValue:
-					l = append(l, fmt.Sprintf("%s e.g. column '%s'", IssueDB[i].Brief, spColName))
-				case ForeignKey:
-					l = append(l, fmt.Sprintf("Column '%s' uses foreign keys which HarbourBridge does not support yet", spColName))
-				case AutoIncrement:
-					l = append(l, fmt.Sprintf("Column '%s' is an autoincrement column. %s", spColName, IssueDB[i].Brief))
-				case Timestamp:
-					// Avoid the confusing "timestamp is mapped to timestamp" message.
-					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", spColName, IssueDB[i].Brief))
-				case Datetime:
-					l = append(l, fmt.Sprintf("Some columns have source DB type 'datetime' which is mapped to Spanner type timestamp e.g. column '%s'. %s", spColName, IssueDB[i].Brief))
-				case Widened:
-					l = append(l, fmt.Sprintf("%s e.g. for column '%s', source DB type %s is mapped to Spanner data type %s", IssueDB[i].Brief, spColName, srcColType, spColType))
-				case HotspotTimestamp:
-					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-				case HotspotAutoIncrement:
-					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-				case InterleavedNotInOrder:
-					parent, _, _ := getInterleaveDetail(conv, tableId, colId, i)
-					str := fmt.Sprintf(" Table %s can be interleaved with table %s %s  %s and Column %s", spSchema.Name, parent, IssueDB[i].Brief, spSchema.Name, spColName)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-				case InterleavedOrder:
-					parent, _, _ := getInterleaveDetail(conv, tableId, colId, i)
-					str := fmt.Sprintf("Table %s %s %s go to Interleave Table Tab", spSchema.Name, IssueDB[i].Brief, parent)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-				case InterleavedAddColumn:
-					parent, _, _ := getInterleaveDetail(conv, tableId, colId, i)
-					str := fmt.Sprintf(" %s %s add %s as a primary key in table %s", IssueDB[i].Brief, parent, spColName, spSchema.Name)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-				case InterleavedRenameColumn:
-					parent, fkName, referColName := getInterleaveDetail(conv, tableId, colId, i)
-					str := fmt.Sprintf(" %s %s rename %s primary key in table %s to match the foreign key %s refer column \"%s\"", IssueDB[i].Brief, parent, spColName, spSchema.Name, fkName, referColName)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-
-				case RedundantIndex:
-					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-
-				case AutoIncrementIndex:
-					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-
-				case InterleaveIndex:
-					str := fmt.Sprintf("Column %s of Table %s %s", spColName, spSchema.Name, IssueDB[i].Brief)
-
-					if !contains(l, str) {
-						l = append(l, str)
-					}
-
-				case IllegalName:
-					l = append(l, fmt.Sprintf("%s, Column '%s' is mapped to '%s'", IssueDB[i].Brief, srcColName, spColName))
-				default:
-					l = append(l, fmt.Sprintf("Column '%s': type %s is mapped to %s. %s", spColName, srcColType, spColType, IssueDB[i].Brief))
-				}
-			}
-		}
-		if len(l) == 0 {
-			continue
-		}
-
-		heading := p.heading
-		if len(l) > 1 {
-			heading = heading + "s"
-		}
-		body = append(body, tableReportBody{Heading: heading, Lines: l})
-	}
-	return body
-}
-
-func getInterleaveDetail(conv *Conv, tableId string, colId string, issueType SchemaIssue) (parent, fkName, referColName string) {
-	table := conv.SpSchema[tableId]
-	for _, fk := range table.ForeignKeys {
-		for i, columnId := range fk.ColIds {
-			if columnId != colId {
-				continue
-			}
-			colPkOrder, err1 := getPkOrderForReport(table.PrimaryKeys, columnId)
-			refColPkOrder, err2 := getPkOrderForReport(conv.SpSchema[fk.ReferTableId].PrimaryKeys, fk.ReferColumnIds[i])
-
-			if err2 != nil || refColPkOrder != 1 {
-				continue
-			}
-
-			switch issueType {
-			case InterleavedOrder:
-				if colPkOrder == 1 && err1 == nil {
-					return conv.SpSchema[fk.ReferTableId].Name, "", ""
-				}
-			case InterleavedNotInOrder:
-				if err1 == nil && colPkOrder != 1 {
-					return conv.SpSchema[fk.ReferTableId].Name, "", ""
-				}
-			case InterleavedRenameColumn:
-				if err1 == nil {
-					parentTable := conv.SpSchema[fk.ReferTableId]
-					return conv.SpSchema[fk.ReferTableId].Name, fk.Name, parentTable.ColDefs[fk.ReferColumnIds[i]].Name
-				}
-			case InterleavedAddColumn:
-				if err1 != nil {
-					return conv.SpSchema[fk.ReferTableId].Name, "", ""
-				}
-			}
-		}
-	}
-	return "", "", ""
-}
-
-func getPkOrderForReport(pks []ddl.IndexKey, colId string) (int, error) {
-	for _, pk := range pks {
-		if pk.ColId == colId {
-			return pk.Order, nil
-		}
-	}
-	return 0, fmt.Errorf("column is not a part of primary key")
-}
-
-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", srcTable, rows, goodConvRows, badConvRows, badRowWrites))
-	}
-	tr.rows = rows
-	tr.badRows = badConvRows + badRowWrites
-}
-
-// IssueDB provides a description and severity for each schema issue.
-// Note on batch: for some issues, we'd like to report just the first instance
-// in a table and suppress other instances i.e. adding more instances
-// of the issue in the same table has little value and could be very noisy.
-// This is controlled via 'batch': if true, we count only the first instance
-// for assessing warnings, and we give only the first instance in the report.
-// TODO: add links in these descriptions to further documentation
-// e.g. for timestamp description.
-var IssueDB = map[SchemaIssue]struct {
-	Brief    string // Short description of issue.
-	severity severity
-	batch    bool // Whether multiple instances of this issue are combined.
-}{
-	DefaultValue:            {Brief: "Some columns have default values which Spanner does not support", severity: warning, batch: true},
-	ForeignKey:              {Brief: "Spanner does not support foreign keys", severity: warning},
-	MultiDimensionalArray:   {Brief: "Spanner doesn't support multi-dimensional arrays", severity: warning},
-	NoGoodType:              {Brief: "No appropriate Spanner type", severity: warning},
-	Numeric:                 {Brief: "Spanner does not support numeric. This type mapping could lose precision and is not recommended for production use", severity: warning},
-	NumericThatFits:         {Brief: "Spanner does not support numeric, but this type mapping preserves the numeric's specified precision", severity: suggestion},
-	Decimal:                 {Brief: "Spanner does not support decimal. This type mapping could lose precision and is not recommended for production use", severity: warning},
-	DecimalThatFits:         {Brief: "Spanner does not support decimal, but this type mapping preserves the decimal's specified precision", severity: suggestion},
-	Serial:                  {Brief: "Spanner does not support autoincrementing types", severity: warning},
-	AutoIncrement:           {Brief: "Spanner does not support auto_increment attribute", severity: warning},
-	Timestamp:               {Brief: "Spanner timestamp is closer to PostgreSQL timestamptz", severity: suggestion, batch: true},
-	Datetime:                {Brief: "Spanner timestamp is closer to MySQL timestamp", severity: warning, batch: true},
-	Time:                    {Brief: "Spanner does not support time/year types", severity: warning, batch: true},
-	Widened:                 {Brief: "Some columns will consume more storage in Spanner", severity: warning, batch: true},
-	StringOverflow:          {Brief: "String overflow issue might occur as maximum supported length in Spanner is 2621440", severity: warning},
-	HotspotTimestamp:        {Brief: "Timestamp Hotspot Occured", severity: warning},
-	HotspotAutoIncrement:    {Brief: "Autoincrement Hotspot Occured", severity: warning},
-	InterleavedOrder:        {Brief: "can be converted as Interleaved with Table", severity: suggestion},
-	RedundantIndex:          {Brief: "Redundant Index", severity: warning},
-	AutoIncrementIndex:      {Brief: "Auto increment column in Index can create a Hotspot", severity: warning},
-	InterleaveIndex:         {Brief: "can be converted to an Interleave Index", severity: suggestion},
-	InterleavedNotInOrder:   {Brief: "if primary key order parameter is changed to 1 for the table", severity: suggestion},
-	InterleavedAddColumn:    {Brief: "Candidate for Interleaved Table", severity: suggestion},
-	IllegalName:             {Brief: "Names must adhere to the spanner regular expression {a-z|A-Z}[{a-z|A-Z|0-9|_}+]", severity: warning},
-	InterleavedRenameColumn: {Brief: "Candidate for Interleaved Table", severity: suggestion},
-}
-
-type severity int
-
-const (
-	warning severity = iota
-	note
-	suggestion
-	errors
-)
-
-// AnalyzeCols returns information about the quality of schema mappings
-// for table 'srcTable'. It assumes 'srcTable' is in the conv.SrcSchema map.
-func AnalyzeCols(conv *Conv, tableId string) (map[string][]SchemaIssue, int64, int64) {
-	srcSchema := conv.SrcSchema[tableId]
-	m := make(map[string][]SchemaIssue)
-	warnings := int64(0)
-	warningBatcher := make(map[SchemaIssue]bool)
-	// Note on how we count warnings when there are multiple warnings
-	// 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, l := range conv.SchemaIssues[tableId] {
-		colWarning := false
-		m[c] = l
-		for _, i := range l {
-			switch {
-			case IssueDB[i].severity == warning && IssueDB[i].batch:
-				warningBatcher[i] = true
-			case IssueDB[i].severity == warning && !IssueDB[i].batch:
-				colWarning = true
-			}
-		}
-		if colWarning {
-			warnings++
-		}
-	}
-	warnings += int64(len(warningBatcher))
-	return m, int64(len(srcSchema.ColDefs)), warnings
-}
-
-// 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 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, string) {
-	pkMsg := "missing primary key"
-	s := fmt.Sprintf(" (%s%% of %d columns mapped cleanly)", pct(cols, warnings), cols)
-	if summary {
-		pkMsg = "some missing primary keys"
-	}
-	switch {
-	case cols == 0:
-		return "NONE", "NONE (no schema found)"
-	case warnings == 0 && !missingPKey:
-		return "EXCELLENT", fmt.Sprintf("EXCELLENT (all %d columns mapped cleanly)", cols)
-	case warnings == 0 && missingPKey:
-		return "GOOD", fmt.Sprintf("GOOD (all columns mapped cleanly, but %s)", pkMsg)
-	case good(cols, warnings) && !missingPKey:
-		return "GOOD", "GOOD" + s
-	case good(cols, warnings) && missingPKey:
-		return "GOOD", "GOOD" + s + fmt.Sprintf(" + %s", pkMsg)
-	case ok(cols, warnings) && !missingPKey:
-		return "OK", "OK" + s
-	case ok(cols, warnings) && missingPKey:
-		return "OK", "OK" + s + fmt.Sprintf(" + %s", pkMsg)
-	case !missingPKey:
-		return "POOR", "POOR" + s
-	default:
-		return "POOR", "POOR" + s + fmt.Sprintf(" + %s", pkMsg)
-	}
-}
-
-func rateData(rows int64, badRows int64, dryRun bool) string {
-	reportText := ""
-	if dryRun {
-		reportText = "successfully converted"
-	} else {
-		reportText = "written"
-	}
-	s := fmt.Sprintf(" (%s%% of %d rows %s to Spanner)", pct(rows, badRows), rows, reportText)
-	switch {
-	case rows == 0:
-		return "NONE (no data rows found)"
-	case badRows == 0:
-		return fmt.Sprintf("EXCELLENT (all %d rows %s to Spanner)", rows, reportText)
-	case good(rows, badRows):
-		return "GOOD" + s
-	case ok(rows, badRows):
-		return "OK" + s
-	default:
-		return "POOR" + s
-	}
-}
-
-func good(total, badCount int64) bool {
-	return float64(badCount) < float64(total)/20
-}
-
-func ok(total, badCount int64) bool {
-	return float64(badCount) < (float64(total))/3
-}
-
-func rateConversion(rows, badRows, cols, warnings int64, missingPKey, summary bool, schemaOnly bool, migrationType migration.MigrationData_MigrationType, dryRun bool) string {
-	rate := ""
-	if migrationType != migration.MigrationData_DATA_ONLY {
-		_, rateSchemaReport := RateSchema(cols, warnings, missingPKey, summary)
-		rate = rate + fmt.Sprintf("Schema conversion: %s.\n", rateSchemaReport)
-	}
-	if !schemaOnly {
-		rate = rate + fmt.Sprintf("Data conversion: %s.\n", rateData(rows, badRows, dryRun))
-	}
-	return rate
-}
-
-// GenerateSummary creates a summarized version of a tableReport.
-func GenerateSummary(conv *Conv, r []tableReport, badWrites map[string]int64) string {
-	cols := int64(0)
-	warnings := int64(0)
-	missingPKey := false
-	for _, t := range r {
-		weight := t.rows // Weight col data by how many rows in table.
-		if weight == 0 { // Tables without data count as if they had one row.
-			weight = 1
-		}
-		cols += t.Cols * weight
-		warnings += t.Warnings * weight
-		if t.SyntheticPKey != "" {
-			missingPKey = true
-		}
-	}
-	// Don't use tableReport for rows/badRows stats because tableReport
-	// provides per-table stats for each table in the schema i.e. it omits
-	// rows for tables not in the schema. To handle this corner-case, use
-	// the source of truth for row stats: conv.Stats.
-	rows := conv.Rows()
-	badRows := conv.BadRows() // Bad rows encountered during data conversion.
-	// Add in bad rows while writing to Spanner.
-	for _, n := range badWrites {
-		badRows += n
-	}
-	return rateConversion(rows, badRows, cols, warnings, missingPKey, true, conv.SchemaMode(), *conv.Audit.MigrationType, conv.Audit.DryRun)
-}
-
-// IgnoredStatements creates a list of statements to ignore.
-func IgnoredStatements(conv *Conv) (l []string) {
-	for s := range conv.Stats.Statement {
-		switch s {
-		case "CreateFunctionStmt":
-			l = append(l, "functions")
-		case "CreateSeqStmt", "CreateSequenceStmt":
-			l = append(l, "sequences")
-		case "CreatePLangStmt", "CreateProcedureStmt":
-			l = append(l, "procedures")
-		case "CreateTrigStmt":
-			l = append(l, "triggers")
-		case "IndexStmt", "CreateIndexStmt":
-			l = append(l, "(non-primary) indexes")
-		case "ViewStmt", "CreateViewStmt":
-			l = append(l, "views")
-		}
-	}
-	sort.Strings(l)
-	return l
-}
-
-func reportNameChanges(conv *Conv, w *bufio.Writer) {
-
-	w.WriteString("-----------------------------------------------------------------------------------------------------\n")
-	w.WriteString("Name Changes in Migration\n")
-	w.WriteString("-----------------------------------------------------------------------------------------------------\n")
-	fmt.Fprintf(w, "%25s %15s %25s %25s\n", "Source Table", "Change", "Old Name", "New Name")
-	w.WriteString("-----------------------------------------------------------------------------------------------------\n")
-
-	for tableId, spTable := range conv.SpSchema {
-		srcTable := conv.SrcSchema[tableId]
-		if srcTable.Name != spTable.Name {
-			fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTable.Name, "Table Name", srcTable.Name, spTable.Name)
-		}
-		for colId, spCol := range spTable.ColDefs {
-			srcCol, ok := srcTable.ColDefs[colId]
-			if !ok {
-				continue
-			}
-			if srcCol.Name != spCol.Name {
-				fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTable.Name, "Column Name", srcCol.Name, spCol.Name)
-			}
-		}
-		for _, spFk := range conv.SpSchema[tableId].ForeignKeys {
-			srcFk, err := GetSrcFkFromId(conv.SrcSchema[tableId].ForeignKeys, spFk.Id)
-			if err != nil {
-				continue
-			}
-			if srcFk.Name != spFk.Name {
-				fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTable.Name, "Foreign Key", srcFk.Name, spFk.Name)
-			}
-		}
-		for _, spIdx := range conv.SpSchema[tableId].Indexes {
-			srcIdx, err := GetSrcIndexFromId(conv.SrcSchema[tableId].Indexes, spIdx.Id)
-			if err != nil {
-				continue
-			}
-			if srcIdx.Name != spIdx.Name {
-				fmt.Fprintf(w, "%25s %15s %25s %25s\n", srcTable.Name, "Index", srcIdx.Name, spIdx.Name)
-			}
-		}
-	}
-	w.WriteString("-----------------------------------------------------------------------------------------------------\n\n\n")
-
-}
-
-func writeStmtStats(driverName string, conv *Conv, w *bufio.Writer) {
-	type stat struct {
-		statement string
-		count     int64
-	}
-	var l []stat
-	for s, x := range conv.Stats.Statement {
-		l = append(l, stat{s, x.Schema + x.Data + x.Skip + x.Error})
-	}
-	// Sort by alphabetical order of statements.
-	sort.Slice(l, func(i, j int) bool {
-		return l[i].statement < l[j].statement
-	})
-	writeHeading(w, "Statements Processed")
-	w.WriteString("Analysis of statements in " + driverName + " output, broken down by statement type.\n")
-	w.WriteString("  schema: statements successfully processed for Spanner schema information.\n")
-	w.WriteString("    data: statements successfully processed for data.\n")
-	w.WriteString("    skip: statements not relevant for Spanner schema or data.\n")
-	w.WriteString("   error: statements that could not be processed.\n")
-	w.WriteString("  --------------------------------------\n")
-	fmt.Fprintf(w, "  %6s %6s %6s %6s  %s\n", "schema", "data", "skip", "error", "statement")
-	w.WriteString("  --------------------------------------\n")
-	for _, x := range l {
-		s := conv.Stats.Statement[x.statement]
-		fmt.Fprintf(w, "  %6d %6d %6d %6d  %s\n", s.Schema, s.Data, s.Skip, s.Error, x.statement)
-	}
-	if driverName == constants.PGDUMP {
-		w.WriteString("See github.com/pganalyze/pg_query_go for definitions of statement types\n")
-		w.WriteString("(pganalyze/pg_query_go is the library we use for parsing pg_dump output).\n")
-		w.WriteString("\n")
-	} else if driverName == constants.MYSQLDUMP {
-		w.WriteString("See https://github.com/pingcap/parser for definitions of statement types\n")
-		w.WriteString("(pingcap/tidb/parser is the library we use for parsing mysqldump output).\n")
-		w.WriteString("\n")
-	}
-}
-
-func writeUnexpectedConditions(driverName string, conv *Conv, w *bufio.Writer) {
-	reparseInfo := func() {
-		if conv.Stats.Reparsed > 0 {
-			fmt.Fprintf(w, "Note: there were %d %s reparse events while looking for statement boundaries.\n\n", conv.Stats.Reparsed, driverName)
-		}
-	}
-	writeHeading(w, "Unexpected Conditions")
-	if len(conv.Stats.Unexpected) == 0 {
-		w.WriteString("There were no unexpected conditions encountered during processing.\n\n")
-		reparseInfo()
-		return
-	}
-	switch driverName {
-	case constants.MYSQLDUMP:
-		w.WriteString("For debugging only. This section provides details of unexpected conditions\n")
-		w.WriteString("encountered as we processed the mysqldump data. In particular, the AST node\n")
-		w.WriteString("representation used by the pingcap/tidb/parser library used for parsing\n")
-		w.WriteString("mysqldump output is highly permissive: almost any construct can appear at\n")
-		w.WriteString("any node in the AST tree. The list details all unexpected nodes and\n")
-		w.WriteString("conditions.\n")
-	case constants.PGDUMP:
-		w.WriteString("For debugging only. This section provides details of unexpected conditions\n")
-		w.WriteString("encountered as we processed the pg_dump data. In particular, the AST node\n")
-		w.WriteString("representation used by the pganalyze/pg_query_go library used for parsing\n")
-		w.WriteString("pg_dump output is highly permissive: almost any construct can appear at\n")
-		w.WriteString("any node in the AST tree. The list details all unexpected nodes and\n")
-		w.WriteString("conditions.\n")
-	default:
-		w.WriteString("For debugging only. This section provides details of unexpected conditions\n")
-		w.WriteString("encountered as we processed the " + driverName + " data. The list details\n")
-		w.WriteString("all unexpected conditions\n")
-	}
-	w.WriteString("  --------------------------------------\n")
-	fmt.Fprintf(w, "  %6s  %s\n", "count", "condition")
-	w.WriteString("  --------------------------------------\n")
-	for s, n := range conv.Stats.Unexpected {
-		fmt.Fprintf(w, "  %6d  %s\n", n, s)
-	}
-	w.WriteString("\n")
-	reparseInfo()
-}
-
-// justifyLines writes s out to w, adding newlines between words
-// to keep line length under 'limit'. Newlines are indented
-// 'indent' spaces.
-func justifyLines(w *bufio.Writer, s string, limit int, indent int) {
-	n := 0
-	startOfLine := true
-	words := strings.Split(s, " ") // This only handles spaces (newlines, tabs ignored).
-	for _, x := range words {
-		if n+len(x) > limit && !startOfLine {
-			w.WriteString("\n")
-			w.WriteString(strings.Repeat(" ", indent))
-			n = indent
-			startOfLine = true
-		}
-		if startOfLine {
-			w.WriteString(x)
-			n += len(x)
-		} else {
-			w.WriteString(" " + x)
-			n += len(x) + 1
-		}
-		startOfLine = false
-	}
-}
-
-// pct prints a percentage representation of (total-bad)/total
-func pct(total, bad int64) string {
-	if bad == 0 || total == 0 {
-		return "100"
-	}
-	pct := 100.0 * float64(total-bad) / float64(total)
-	if pct > 99.9 {
-		return fmt.Sprintf("%2.5f", pct)
-	}
-	if pct > 95.0 {
-		return fmt.Sprintf("%2.3f", pct)
-	}
-	return fmt.Sprintf("%2.0f", pct)
-}
-
-func writeHeading(w *bufio.Writer, s string) {
-	w.WriteString(strings.Join([]string{
-		"----------------------------\n",
-		s, "\n",
-		"----------------------------\n"}, ""))
-}
-
-func conversionDuration(conv *Conv, w *bufio.Writer) string {
-	res := ""
-	if conv.Audit.DataConversionDuration.Microseconds() != 0 || conv.Audit.SchemaConversionDuration.Microseconds() != 0 {
-		writeHeading(w, "Time duration of Conversion")
-		if conv.Audit.SchemaConversionDuration.Microseconds() != 0 {
-			res += fmt.Sprintf("Schema conversion duration : %s \n", conv.Audit.SchemaConversionDuration)
-		}
-		if conv.Audit.DataConversionDuration.Microseconds() != 0 {
-			res += fmt.Sprintf("Data conversion duration : %s \n", conv.Audit.DataConversionDuration)
-		}
-		res += "\n"
-	}
-	return res
-}
diff --git a/internal/reports/REPORT.md b/internal/reports/REPORT.md
new file mode 100644
index 0000000..070e68e
--- /dev/null
+++ b/internal/reports/REPORT.md
@@ -0,0 +1,26 @@
+## Harbourbridge Report
+A Harbourbridge report consists of elements defined below. A sample structured report can be found [here](/test_data/mysql_structured_report.json).
+
+#### Summary
+Defines the overall quality of the conversion perfomed by Harbourbridge along with a rating.
+
+#### Migration Type
+Defines the type of conversion performed by Harbourbridge. It is one of SCHEMA, DATA and SCHEMA_AND_DATA.
+
+#### Ignored Statements
+Defines the statements in the source schema which have been ignored by Harbourbridge. For example, View related statements are currently ignored by Harbourbridge.
+#### Conversion Metdata
+Defines the total time taken to perform the conversion. This may include other conversion related metadata in the future.
+
+#### Statement Stats
+Statistics on different types of statements identified by Harbourbridge. This is only populated when processing dump files.
+
+#### Name Changes
+Renaming related changes done by Harbourbridge to ensure Cloud Spanner compatibility.
+
+#### Individual Table Reports
+Detailed table-by-table analysis showing how many columns were converted perfectly, with warnings etc.
+
+#### Unexpected Conditions
+Unexpected conditions encountered by Harbourbridge while processing the source schema/data.
+
diff --git a/internal/reports/report_helpers.go b/internal/reports/report_helpers.go
new file mode 100644
index 0000000..ad4c0cd
--- /dev/null
+++ b/internal/reports/report_helpers.go
@@ -0,0 +1,542 @@
+// Copyright 2023 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 reports
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
+	"github.com/cloudspannerecosystem/harbourbridge/schema"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
+)
+
+//report_helpers.go contains helpers methods to calculate the various elements of a report.
+//Calculation of new elements should go here.
+
+type tableReport struct {
+	SrcTable      string
+	SpTable       string
+	rows          int64
+	badRows       int64
+	Cols          int64
+	Warnings      int64
+	SyntheticPKey string // Empty string means no synthetic primary key was needed.
+	Body          []tableReportBody
+}
+
+type tableReportBody struct {
+	Heading string
+	Lines   []string
+}
+
+// AnalyzeTables generates table reports for all processed tables.
+func AnalyzeTables(conv *internal.Conv, badWrites map[string]int64) (r []tableReport) {
+	// Process tables in alphabetical order. This ensures that tables
+	// appear in alphabetical order in report.txt.
+	var tableNames []string
+	for _, srcTable := range conv.SrcSchema {
+		tableNames = append(tableNames, srcTable.Name)
+	}
+	sort.Strings(tableNames)
+	for _, tableName := range tableNames {
+		tableId, err := internal.GetTableIdFromSrcName(conv.SrcSchema, tableName)
+		if err != nil {
+			continue
+		}
+		if _, isPresent := conv.SpSchema[tableId]; isPresent {
+			r = append(r, buildTableReport(conv, tableId, badWrites))
+		}
+	}
+	return r
+}
+
+func buildTableReport(conv *internal.Conv, tableId string, badWrites map[string]int64) tableReport {
+	srcSchema, ok1 := conv.SrcSchema[tableId]
+	spSchema, ok2 := conv.SpSchema[tableId]
+	tr := tableReport{SrcTable: tableId, SpTable: tableId}
+	if !ok1 || !ok2 {
+		m := "bad source-DB-to-Spanner table mapping or Spanner schema"
+		conv.Unexpected("report: " + m)
+		tr.Body = []tableReportBody{{Heading: "Internal error: " + m}}
+		return tr
+	}
+	if *conv.Audit.MigrationType != migration.MigrationData_DATA_ONLY {
+		issues, cols, warnings := AnalyzeCols(conv, tableId)
+		tr.Cols = cols
+		tr.Warnings = warnings
+		if pk, ok := conv.SyntheticPKeys[tableId]; ok {
+			tr.SyntheticPKey = pk.ColId
+			synthColName := conv.SpSchema[tableId].ColDefs[pk.ColId].Name
+			tr.Body = buildTableReportBody(conv, tableId, issues, spSchema, srcSchema, &synthColName, nil)
+		} else if pk, ok := conv.UniquePKey[tableId]; ok {
+			tr.Body = buildTableReportBody(conv, tableId, issues, spSchema, srcSchema, nil, pk)
+		} else {
+			tr.Body = buildTableReportBody(conv, tableId, issues, spSchema, srcSchema, nil, nil)
+		}
+
+	}
+	if !conv.SchemaMode() {
+		fillRowStats(conv, tableId, badWrites, &tr)
+	}
+	return tr
+}
+
+func buildTableReportBody(conv *internal.Conv, tableId string, issues map[string][]internal.SchemaIssue, spSchema ddl.CreateTable, srcSchema schema.Table, syntheticPK *string, uniquePK []string) []tableReportBody {
+	var body []tableReportBody
+	for _, p := range []struct {
+		heading  string
+		severity severity
+	}{
+		{"Warning", warning},
+		{"Note", note},
+		{"Suggestion", suggestion},
+		{"Error", errors},
+	} {
+		// Print out issues is alphabetical column order.
+		var colNames []string
+		for colId := range issues {
+			colNames = append(colNames, conv.SpSchema[tableId].ColDefs[colId].Name)
+		}
+		sort.Strings(colNames)
+		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 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))
+			}
+		}
+		if uniquePK != nil {
+			// Warning about using a column with unique constraint as primary key
+			// in case primary key is absent.
+			if p.severity == warning {
+				l = append(l, fmt.Sprintf("UNIQUE constraint on column(s) '%s' replaced with primary key since this table didn't have one. Spanner requires a primary key for every table", strings.Join(uniquePK, ", ")))
+			}
+		}
+
+		if p.severity == warning {
+			for _, spFk := range conv.SpSchema[tableId].ForeignKeys {
+				srcFk, err := internal.GetSrcFkFromId(conv.SrcSchema[tableId].ForeignKeys, spFk.Id)
+				if err != nil {
+					continue
+				}
+				_, isChanged := internal.FixName(srcFk.Name)
+				if isChanged && srcFk.Name != spFk.Name {
+					l = append(l, fmt.Sprintf("%s, Foreign Key '%s' is mapped to '%s'", IssueDB[internal.IllegalName].Brief, srcFk.Name, spFk.Name))
+				}
+			}
+			for _, spIdx := range conv.SpSchema[tableId].Indexes {
+				srcIdx, err := internal.GetSrcIndexFromId(conv.SrcSchema[tableId].Indexes, spIdx.Id)
+				if err != nil {
+					continue
+				}
+				_, isChanged := internal.FixName(srcIdx.Name)
+				if isChanged && srcIdx.Name != spIdx.Name {
+					l = append(l, fmt.Sprintf("%s, Index '%s' is mapped to '%s'", IssueDB[internal.IllegalName].Brief, srcIdx.Name, spIdx.Name))
+				}
+			}
+
+			_, isChanged := internal.FixName(srcSchema.Name)
+			if isChanged && (spSchema.Name != srcSchema.Name) {
+				l = append(l, fmt.Sprintf("%s, Table '%s' is mapped to '%s'", IssueDB[internal.IllegalName].Brief, srcSchema.Name, spSchema.Name))
+			}
+		}
+
+		issueBatcher := make(map[internal.SchemaIssue]bool)
+		for _, colName := range colNames {
+			colId, _ := internal.GetColIdFromSpName(conv.SpSchema[tableId].ColDefs, colName)
+			for _, i := range issues[colId] {
+				if IssueDB[i].severity != p.severity {
+					continue
+				}
+				if IssueDB[i].batch {
+					if issueBatcher[i] {
+						// Have already reported a previous instance of this
+						// (batched) issue, so skip this one.
+						continue
+					}
+					issueBatcher[i] = true
+				}
+				srcColType := srcSchema.ColDefs[colId].Type.Print()
+				spColType := spSchema.ColDefs[colId].T.PrintColumnDefType()
+				if conv.SpDialect == constants.DIALECT_POSTGRESQL {
+					spColType = spSchema.ColDefs[colId].T.PGPrintColumnDefType()
+				}
+				srcColName := srcSchema.ColDefs[colId].Name
+				spColName := spSchema.ColDefs[colId].Name
+
+				// 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.
+				// TODO: add logic to choose case for Spanner types based
+				// on case of srcType.
+				spColType = strings.ToLower(spColType)
+				switch i {
+				case internal.DefaultValue:
+					l = append(l, fmt.Sprintf("%s e.g. column '%s'", IssueDB[i].Brief, spColName))
+				case internal.ForeignKey:
+					l = append(l, fmt.Sprintf("Column '%s' uses foreign keys which HarbourBridge does not support yet", spColName))
+				case internal.AutoIncrement:
+					l = append(l, fmt.Sprintf("Column '%s' is an autoincrement column. %s", spColName, IssueDB[i].Brief))
+				case internal.Timestamp:
+					// Avoid the confusing "timestamp is mapped to timestamp" message.
+					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", spColName, IssueDB[i].Brief))
+				case internal.Datetime:
+					l = append(l, fmt.Sprintf("Some columns have source DB type 'datetime' which is mapped to Spanner type timestamp e.g. column '%s'. %s", spColName, IssueDB[i].Brief))
+				case internal.Widened:
+					l = append(l, fmt.Sprintf("%s e.g. for column '%s', source DB type %s is mapped to Spanner data type %s", IssueDB[i].Brief, spColName, srcColType, spColType))
+				case internal.HotspotTimestamp:
+					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+				case internal.HotspotAutoIncrement:
+					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+				case internal.InterleavedNotInOrder:
+					parent, _, _ := getInterleaveDetail(conv, tableId, colId, i)
+					str := fmt.Sprintf(" Table %s can be interleaved with table %s %s  %s and Column %s", spSchema.Name, parent, IssueDB[i].Brief, spSchema.Name, spColName)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+				case internal.InterleavedOrder:
+					parent, _, _ := getInterleaveDetail(conv, tableId, colId, i)
+					str := fmt.Sprintf("Table %s %s %s go to Interleave Table Tab", spSchema.Name, IssueDB[i].Brief, parent)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+				case internal.InterleavedAddColumn:
+					parent, _, _ := getInterleaveDetail(conv, tableId, colId, i)
+					str := fmt.Sprintf(" %s %s add %s as a primary key in table %s", IssueDB[i].Brief, parent, spColName, spSchema.Name)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+				case internal.InterleavedRenameColumn:
+					parent, fkName, referColName := getInterleaveDetail(conv, tableId, colId, i)
+					str := fmt.Sprintf(" %s %s rename %s primary key in table %s to match the foreign key %s refer column \"%s\"", IssueDB[i].Brief, parent, spColName, spSchema.Name, fkName, referColName)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+
+				case internal.RedundantIndex:
+					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+
+				case internal.AutoIncrementIndex:
+					str := fmt.Sprintf(" %s for Table %s and Column  %s", IssueDB[i].Brief, spSchema.Name, spColName)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+
+				case internal.InterleaveIndex:
+					str := fmt.Sprintf("Column %s of Table %s %s", spColName, spSchema.Name, IssueDB[i].Brief)
+
+					if !internal.Contains(l, str) {
+						l = append(l, str)
+					}
+
+				case internal.IllegalName:
+					l = append(l, fmt.Sprintf("%s, Column '%s' is mapped to '%s'", IssueDB[i].Brief, srcColName, spColName))
+				default:
+					l = append(l, fmt.Sprintf("Column '%s': type %s is mapped to %s. %s", spColName, srcColType, spColType, IssueDB[i].Brief))
+				}
+			}
+		}
+		if len(l) == 0 {
+			continue
+		}
+
+		heading := p.heading
+		if len(l) > 1 {
+			heading = heading + "s"
+		}
+		body = append(body, tableReportBody{Heading: heading, Lines: l})
+	}
+	return body
+}
+
+func getInterleaveDetail(conv *internal.Conv, tableId string, colId string, issueType internal.SchemaIssue) (parent, fkName, referColName string) {
+	table := conv.SpSchema[tableId]
+	for _, fk := range table.ForeignKeys {
+		for i, columnId := range fk.ColIds {
+			if columnId != colId {
+				continue
+			}
+			colPkOrder, err1 := getPkOrderForReport(table.PrimaryKeys, columnId)
+			refColPkOrder, err2 := getPkOrderForReport(conv.SpSchema[fk.ReferTableId].PrimaryKeys, fk.ReferColumnIds[i])
+
+			if err2 != nil || refColPkOrder != 1 {
+				continue
+			}
+
+			switch issueType {
+			case internal.InterleavedOrder:
+				if colPkOrder == 1 && err1 == nil {
+					return conv.SpSchema[fk.ReferTableId].Name, "", ""
+				}
+			case internal.InterleavedNotInOrder:
+				if err1 == nil && colPkOrder != 1 {
+					return conv.SpSchema[fk.ReferTableId].Name, "", ""
+				}
+			case internal.InterleavedRenameColumn:
+				if err1 == nil {
+					parentTable := conv.SpSchema[fk.ReferTableId]
+					return conv.SpSchema[fk.ReferTableId].Name, fk.Name, parentTable.ColDefs[fk.ReferColumnIds[i]].Name
+				}
+			case internal.InterleavedAddColumn:
+				if err1 != nil {
+					return conv.SpSchema[fk.ReferTableId].Name, "", ""
+				}
+			}
+		}
+	}
+	return "", "", ""
+}
+
+func getPkOrderForReport(pks []ddl.IndexKey, colId string) (int, error) {
+	for _, pk := range pks {
+		if pk.ColId == colId {
+			return pk.Order, nil
+		}
+	}
+	return 0, fmt.Errorf("column is not a part of primary key")
+}
+
+func fillRowStats(conv *internal.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", srcTable, rows, goodConvRows, badConvRows, badRowWrites))
+	}
+	tr.rows = rows
+	tr.badRows = badConvRows + badRowWrites
+}
+
+// IssueDB provides a description and severity for each schema issue.
+// Note on batch: for some issues, we'd like to report just the first instance
+// in a table and suppress other instances i.e. adding more instances
+// of the issue in the same table has little value and could be very noisy.
+// This is controlled via 'batch': if true, we count only the first instance
+// for assessing warnings, and we give only the first instance in the report.
+// TODO: add links in these descriptions to further documentation
+// e.g. for timestamp description.
+var IssueDB = map[internal.SchemaIssue]struct {
+	Brief    string // Short description of issue.
+	severity severity
+	batch    bool // Whether multiple instances of this issue are combined.
+}{
+	internal.DefaultValue:            {Brief: "Some columns have default values which Spanner does not support", severity: warning, batch: true},
+	internal.ForeignKey:              {Brief: "Spanner does not support foreign keys", severity: warning},
+	internal.MultiDimensionalArray:   {Brief: "Spanner doesn't support multi-dimensional arrays", severity: warning},
+	internal.NoGoodType:              {Brief: "No appropriate Spanner type", severity: warning},
+	internal.Numeric:                 {Brief: "Spanner does not support numeric. This type mapping could lose precision and is not recommended for production use", severity: warning},
+	internal.NumericThatFits:         {Brief: "Spanner does not support numeric, but this type mapping preserves the numeric's specified precision", severity: suggestion},
+	internal.Decimal:                 {Brief: "Spanner does not support decimal. This type mapping could lose precision and is not recommended for production use", severity: warning},
+	internal.DecimalThatFits:         {Brief: "Spanner does not support decimal, but this type mapping preserves the decimal's specified precision", severity: suggestion},
+	internal.Serial:                  {Brief: "Spanner does not support autoincrementing types", severity: warning},
+	internal.AutoIncrement:           {Brief: "Spanner does not support auto_increment attribute", severity: warning},
+	internal.Timestamp:               {Brief: "Spanner timestamp is closer to PostgreSQL timestamptz", severity: suggestion, batch: true},
+	internal.Datetime:                {Brief: "Spanner timestamp is closer to MySQL timestamp", severity: warning, batch: true},
+	internal.Time:                    {Brief: "Spanner does not support time/year types", severity: warning, batch: true},
+	internal.Widened:                 {Brief: "Some columns will consume more storage in Spanner", severity: warning, batch: true},
+	internal.StringOverflow:          {Brief: "String overflow issue might occur as maximum supported length in Spanner is 2621440", severity: warning},
+	internal.HotspotTimestamp:        {Brief: "Timestamp Hotspot Occured", severity: warning},
+	internal.HotspotAutoIncrement:    {Brief: "Autoincrement Hotspot Occured", severity: warning},
+	internal.InterleavedOrder:        {Brief: "can be converted as Interleaved with Table", severity: suggestion},
+	internal.RedundantIndex:          {Brief: "Redundant Index", severity: warning},
+	internal.AutoIncrementIndex:      {Brief: "Auto increment column in Index can create a Hotspot", severity: warning},
+	internal.InterleaveIndex:         {Brief: "can be converted to an Interleave Index", severity: suggestion},
+	internal.InterleavedNotInOrder:   {Brief: "if primary key order parameter is changed to 1 for the table", severity: suggestion},
+	internal.InterleavedAddColumn:    {Brief: "Candidate for Interleaved Table", severity: suggestion},
+	internal.IllegalName:             {Brief: "Names must adhere to the spanner regular expression {a-z|A-Z}[{a-z|A-Z|0-9|_}+]", severity: warning},
+	internal.InterleavedRenameColumn: {Brief: "Candidate for Interleaved Table", severity: suggestion},
+}
+
+type severity int
+
+const (
+	warning severity = iota
+	note
+	suggestion
+	errors
+)
+
+// AnalyzeCols returns information about the quality of schema mappings
+// for table 'srcTable'. It assumes 'srcTable' is in the conv.SrcSchema map.
+func AnalyzeCols(conv *internal.Conv, tableId string) (map[string][]internal.SchemaIssue, int64, int64) {
+	srcSchema := conv.SrcSchema[tableId]
+	m := make(map[string][]internal.SchemaIssue)
+	warnings := int64(0)
+	warningBatcher := make(map[internal.SchemaIssue]bool)
+	// Note on how we count warnings when there are multiple warnings
+	// 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, l := range conv.SchemaIssues[tableId] {
+		colWarning := false
+		m[c] = l
+		for _, i := range l {
+			switch {
+			case IssueDB[i].severity == warning && IssueDB[i].batch:
+				warningBatcher[i] = true
+			case IssueDB[i].severity == warning && !IssueDB[i].batch:
+				colWarning = true
+			}
+		}
+		if colWarning {
+			warnings++
+		}
+	}
+	warnings += int64(len(warningBatcher))
+	return m, int64(len(srcSchema.ColDefs)), warnings
+}
+
+// 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 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, string) {
+	pkMsg := "missing primary key"
+	s := fmt.Sprintf(" (%s%% of %d columns mapped cleanly)", pct(cols, warnings), cols)
+	if summary {
+		pkMsg = "some missing primary keys"
+	}
+	switch {
+	case cols == 0:
+		return "NONE", "NONE (no schema found)"
+	case warnings == 0 && !missingPKey:
+		return "EXCELLENT", fmt.Sprintf("EXCELLENT (all %d columns mapped cleanly)", cols)
+	case warnings == 0 && missingPKey:
+		return "GOOD", fmt.Sprintf("GOOD (all columns mapped cleanly, but %s)", pkMsg)
+	case good(cols, warnings) && !missingPKey:
+		return "GOOD", "GOOD" + s
+	case good(cols, warnings) && missingPKey:
+		return "GOOD", "GOOD" + s + fmt.Sprintf(" + %s", pkMsg)
+	case ok(cols, warnings) && !missingPKey:
+		return "OK", "OK" + s
+	case ok(cols, warnings) && missingPKey:
+		return "OK", "OK" + s + fmt.Sprintf(" + %s", pkMsg)
+	case !missingPKey:
+		return "POOR", "POOR" + s
+	default:
+		return "POOR", "POOR" + s + fmt.Sprintf(" + %s", pkMsg)
+	}
+}
+
+func rateData(rows int64, badRows int64, dryRun bool) (string, string) {
+	reportText := ""
+	if dryRun {
+		reportText = "successfully converted"
+	} else {
+		reportText = "written"
+	}
+	s := fmt.Sprintf(" (%s%% of %d rows %s to Spanner)", pct(rows, badRows), rows, reportText)
+	switch {
+	case rows == 0:
+		return "NONE", "NONE (no data rows found)"
+	case badRows == 0:
+		return "EXCELLENT", fmt.Sprintf("EXCELLENT (all %d rows %s to Spanner)", rows, reportText)
+	case good(rows, badRows):
+		return "GOOD", "GOOD" + s
+	case ok(rows, badRows):
+		return "OK", "OK" + s
+	default:
+		return "POOR", "POOR" + s
+	}
+}
+
+func good(total, badCount int64) bool {
+	return float64(badCount) < float64(total)/20
+}
+
+func ok(total, badCount int64) bool {
+	return float64(badCount) < (float64(total))/3
+}
+
+func rateConversion(rows, badRows, cols, warnings int64, missingPKey, summary bool, schemaOnly bool, migrationType migration.MigrationData_MigrationType, dryRun bool) (string, string) {
+	rate := ""
+	var rating string
+	if migrationType != migration.MigrationData_DATA_ONLY {
+		var rateSchemaReport string
+		rating, rateSchemaReport = RateSchema(cols, warnings, missingPKey, summary)
+		rate = rate + fmt.Sprintf("Schema conversion: %s.\n", rateSchemaReport)
+	}
+	if !schemaOnly {
+		var rateDataReport string
+		rating, rateDataReport = rateData(rows, badRows, dryRun)
+		rate = rate + fmt.Sprintf("Data conversion: %s.\n", rateDataReport)
+	}
+	return rating, rate
+}
+
+// GenerateSummary creates a summarized version of a tableReport.
+func GenerateSummary(conv *internal.Conv, r []tableReport, badWrites map[string]int64) (string, string) {
+	cols := int64(0)
+	warnings := int64(0)
+	missingPKey := false
+	for _, t := range r {
+		weight := t.rows // Weight col data by how many rows in table.
+		if weight == 0 { // Tables without data count as if they had one row.
+			weight = 1
+		}
+		cols += t.Cols * weight
+		warnings += t.Warnings * weight
+		if t.SyntheticPKey != "" {
+			missingPKey = true
+		}
+	}
+	// Don't use tableReport for rows/badRows stats because tableReport
+	// provides per-table stats for each table in the schema i.e. it omits
+	// rows for tables not in the schema. To handle this corner-case, use
+	// the source of truth for row stats: conv.Stats.
+	rows := conv.Rows()
+	badRows := conv.BadRows() // Bad rows encountered during data conversion.
+	// Add in bad rows while writing to Spanner.
+	for _, n := range badWrites {
+		badRows += n
+	}
+	return rateConversion(rows, badRows, cols, warnings, missingPKey, true, conv.SchemaMode(), *conv.Audit.MigrationType, conv.Audit.DryRun)
+}
diff --git a/internal/reports/report_text.go b/internal/reports/report_text.go
new file mode 100644
index 0000000..820636d
--- /dev/null
+++ b/internal/reports/report_text.go
@@ -0,0 +1,285 @@
+package reports
+
+import (
+	"bufio"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+)
+
+//report_text.go contains the logic to convert a structured harbourbridge 
+//report to a human readable text report.
+// The structure of the report created is present in (internal/reports/REPORT.md)
+// A sample report can be found in (test_data/mysql_text_report.txt)
+func GenerateTextReport(structuredReport StructuredReport, w *bufio.Writer) {
+	writeHeading(w, "Summary of Conversion")
+	w.WriteString(structuredReport.Summary.Text)
+	w.WriteString("\n")
+	w.WriteString(writeConversionMetadata(structuredReport.ConversionMetadata, w))
+	if len(structuredReport.IgnoredStatements) > 0 {
+		justifyLines(w, fmt.Sprintf("Note that the following source DB statements "+
+			"were detected but ignored: %s.",
+			strings.Join(getStatementsFromIgnoredStatements(structuredReport.IgnoredStatements), ", ")), 80, 0)
+		w.WriteString("\n\n")
+	}
+	statementsMsg := ""
+	var isDump bool
+	if strings.Contains(structuredReport.StatementStats.DriverName, "dump") {
+		isDump = true
+	}
+	if isDump {
+		statementsMsg = "stats on the " + structuredReport.StatementStats.DriverName + " statements processed, followed by "
+	}
+	justifyLines(w, "The remainder of this report provides "+statementsMsg+
+		"a table-by-table listing of "+structuredReport.MigrationType+" conversion details. "+
+		"For background on the "+structuredReport.MigrationType+" conversion process used, "+
+		"and explanations of the terms and notes used in this "+
+		"report, see HarbourBridge's README.", 80, 0)
+	w.WriteString("\n\n")
+	if isDump {
+		writeStatementStats(structuredReport, w)
+	}
+	writeNameChanges(structuredReport, w)
+	writeTableReports(structuredReport, w)
+	writeUnexpectedConditionsv2(structuredReport, w)
+
+}
+
+func writeUnexpectedConditionsv2(structuredReport StructuredReport, w *bufio.Writer) {
+	reparseInfo := func() {
+		if structuredReport.UnexpectedConditions.Reparsed > 0 {
+			fmt.Fprintf(w, "Note: there were %d %s reparse events while looking for statement boundaries.\n\n", structuredReport.UnexpectedConditions.Reparsed, structuredReport.StatementStats.DriverName)
+		}
+	}
+	writeHeading(w, "Unexpected Conditions")
+	if len(structuredReport.UnexpectedConditions.UnexpectedConditions) == 0 {
+		w.WriteString("There were no unexpected conditions encountered during processing.\n\n")
+		reparseInfo()
+		return
+	}
+	switch structuredReport.StatementStats.DriverName {
+	case constants.MYSQLDUMP:
+		w.WriteString("For debugging only. This section provides details of unexpected conditions\n")
+		w.WriteString("encountered as we processed the mysqldump data. In particular, the AST node\n")
+		w.WriteString("representation used by the pingcap/tidb/parser library used for parsing\n")
+		w.WriteString("mysqldump output is highly permissive: almost any construct can appear at\n")
+		w.WriteString("any node in the AST tree. The list details all unexpected nodes and\n")
+		w.WriteString("conditions.\n")
+	case constants.PGDUMP:
+		w.WriteString("For debugging only. This section provides details of unexpected conditions\n")
+		w.WriteString("encountered as we processed the pg_dump data. In particular, the AST node\n")
+		w.WriteString("representation used by the pganalyze/pg_query_go library used for parsing\n")
+		w.WriteString("pg_dump output is highly permissive: almost any construct can appear at\n")
+		w.WriteString("any node in the AST tree. The list details all unexpected nodes and\n")
+		w.WriteString("conditions.\n")
+	default:
+		w.WriteString("For debugging only. This section provides details of unexpected conditions\n")
+		w.WriteString("encountered as we processed the " + structuredReport.StatementStats.DriverName + " data. The list details\n")
+		w.WriteString("all unexpected conditions\n")
+	}
+	w.WriteString("  --------------------------------------\n")
+	fmt.Fprintf(w, "  %6s  %s\n", "count", "condition")
+	w.WriteString("  --------------------------------------\n")
+	for _, cond := range structuredReport.UnexpectedConditions.UnexpectedConditions {
+		fmt.Fprintf(w, "  %6d  %s\n", cond.Count, cond.Condition)
+	}
+	w.WriteString("\n")
+	reparseInfo()
+}
+
+// Generates table by table report from the structured report in a text based format.
+// This looks like the following -
+// ----------------------------
+// Table no_pk
+// ----------------------------
+// Schema conversion: POOR (67% of 3 columns mapped cleanly) + missing primary key.
+// Data conversion: POOR (60% of 5000 rows written to Spanner).
+
+// Warnings
+// 1) Column 'synth_id' was added because this table didn't have a primary key.
+//    Spanner requires a primary key for every table.
+// 2) Some columns will consume more storage in Spanner e.g. for column 'b', source
+//    DB type int(11) is mapped to Spanner data type int64.
+func writeTableReports(structuredReport StructuredReport, w *bufio.Writer) {
+	for _, tableReport := range structuredReport.TableReports {
+		h := fmt.Sprintf("Table %s", tableReport.SrcTableName)
+		if tableReport.SrcTableName != tableReport.SpTableName {
+			h = h + fmt.Sprintf(" (mapped to Spanner table %s)", tableReport.SpTableName)
+		}
+		writeHeading(w, h)
+		rate := ""
+		if structuredReport.MigrationType == "SCHEMA" || structuredReport.MigrationType == "SCHEMA_AND_DATA" {
+			schemaRatingText := ""
+			pkMsg := " missing primary key"
+			s := fmt.Sprintf(" (%s%% of %d columns mapped cleanly)", pct(tableReport.SchemaReport.TotalColumns, tableReport.SchemaReport.Warnings), tableReport.SchemaReport.TotalColumns)
+			schemaRatingText = schemaRatingText + tableReport.SchemaReport.Rating + s
+			if tableReport.SchemaReport.PkMissing {
+				schemaRatingText = schemaRatingText + fmt.Sprintf(" +%s", pkMsg)
+			}
+			rate = rate + fmt.Sprintf("Schema conversion: %s.\n", schemaRatingText)
+		}
+		if !structuredReport.SchemaOnly {
+			dataRatingText := ""
+			if tableReport.DataReport.DryRun {
+				dataRatingText = "successfully converted"
+			} else {
+				dataRatingText = "written"
+			}
+			s := fmt.Sprintf(" (%s%% of %d rows %s to Spanner)", pct(tableReport.DataReport.TotalRows, tableReport.DataReport.BadRows), tableReport.DataReport.TotalRows, dataRatingText)
+			dataRatingText = tableReport.DataReport.Rating + s
+			rate = rate + fmt.Sprintf("Data conversion: %s.\n", dataRatingText)
+		}
+		w.WriteString(rate)
+		w.WriteString("\n")
+		for _, warning := range tableReport.Warnings {
+			fmt.Fprintf(w, "%s\n", warning.WarningType)
+			for i, l := range warning.WarningList {
+				justifyLines(w, fmt.Sprintf("%d) %s.\n", i+1, l), 80, 3)
+			}
+			w.WriteString("\n")
+		}
+	}
+}
+
+func writeNameChanges(structuredReport StructuredReport, w *bufio.Writer) {
+	if structuredReport.NameChanges != nil {
+		w.WriteString("-----------------------------------------------------------------------------------------------------\n")
+		w.WriteString("Name Changes in Migration\n")
+		w.WriteString("-----------------------------------------------------------------------------------------------------\n")
+		fmt.Fprintf(w, "%25s %15s %25s %25s\n", "Source Table", "Change", "Old Name", "New Name")
+		w.WriteString("-----------------------------------------------------------------------------------------------------\n")
+		for _, nameChange := range structuredReport.NameChanges {
+			fmt.Fprintf(w, "%25s %15s %25s %25s\n", nameChange.SourceTable, nameChange.NameChangeType, nameChange.OldName, nameChange.NewName)
+		}
+		w.WriteString("-----------------------------------------------------------------------------------------------------\n\n\n")
+	} else {
+		w.WriteString("No Name Changes in Migration\n")
+	}
+}
+
+func writeStatementStats(structuredReport StructuredReport, w *bufio.Writer) {
+	type stat struct {
+		statement string
+		count     int64
+	}
+	var l []stat
+	for _, x := range structuredReport.StatementStats.StatementStats {
+		l = append(l, stat{x.Statement, x.Schema + x.Data + x.Skip + x.Error})
+	}
+	// Sort by alphabetical order of statements.
+	sort.Slice(l, func(i, j int) bool {
+		return l[i].statement < l[j].statement
+	})
+	writeHeading(w, "Statements Processed")
+	w.WriteString("Analysis of statements in " + structuredReport.StatementStats.DriverName + " output, broken down by statement type.\n")
+	w.WriteString("  schema: statements successfully processed for Spanner schema information.\n")
+	w.WriteString("    data: statements successfully processed for data.\n")
+	w.WriteString("    skip: statements not relevant for Spanner schema or data.\n")
+	w.WriteString("   error: statements that could not be processed.\n")
+	w.WriteString("  --------------------------------------\n")
+	fmt.Fprintf(w, "  %6s %6s %6s %6s  %s\n", "schema", "data", "skip", "error", "statement")
+	w.WriteString("  --------------------------------------\n")
+	for _, x := range l {
+		s := findStatementStat(structuredReport.StatementStats.StatementStats, x.statement)
+		fmt.Fprintf(w, "  %6d %6d %6d %6d  %s\n", s.Schema, s.Data, s.Skip, s.Error, x.statement)
+	}
+	if structuredReport.StatementStats.DriverName == constants.PGDUMP {
+		w.WriteString("See github.com/pganalyze/pg_query_go for definitions of statement types\n")
+		w.WriteString("(pganalyze/pg_query_go is the library we use for parsing pg_dump output).\n")
+		w.WriteString("\n")
+	} else if structuredReport.StatementStats.DriverName == constants.MYSQLDUMP {
+		w.WriteString("See https://github.com/pingcap/parser for definitions of statement types\n")
+		w.WriteString("(pingcap/tidb/parser is the library we use for parsing mysqldump output).\n")
+		w.WriteString("\n")
+	}
+}
+
+func findStatementStat(statementStats []StatementStat, statement string) StatementStat {
+	for _, statementStat := range statementStats {
+		if statementStat.Statement == statement {
+			return statementStat
+		}
+	}
+	return StatementStat{}
+}
+
+func writeConversionMetadata(ConversionMetadataList []ConversionMetadata, w *bufio.Writer) string {
+	var schemaConvMetadata, dataConvMetadata ConversionMetadata
+	for _, convMetadata := range ConversionMetadataList {
+		if convMetadata.ConversionType == "Schema" && convMetadata.Duration.Microseconds() != 0 {
+			schemaConvMetadata = convMetadata
+		}
+		if convMetadata.ConversionType == "Data" && convMetadata.Duration.Microseconds() != 0 {
+			dataConvMetadata = convMetadata
+		}
+	}
+	res := ""
+	if schemaConvMetadata.Duration.Microseconds() != 0 || dataConvMetadata.Duration.Microseconds() != 0 {
+		writeHeading(w, "Time duration of Conversion")
+		if schemaConvMetadata.Duration.Microseconds() != 0 {
+			res += fmt.Sprintf("Schema conversion duration : %s \n", schemaConvMetadata.Duration)
+		}
+		if dataConvMetadata.Duration.Microseconds() != 0 {
+			res += fmt.Sprintf("Data conversion duration : %s \n", dataConvMetadata.Duration)
+		}
+		res += "\n"
+	}
+	return res
+}
+
+func getStatementsFromIgnoredStatements(ignoredStatements []IgnoredStatement) []string {
+	var statements []string
+	for _, ignoredStatment := range ignoredStatements {
+		statements = append(statements, ignoredStatment.Statement)
+	}
+	return statements
+}
+
+func writeHeading(w *bufio.Writer, s string) {
+	w.WriteString(strings.Join([]string{
+		"----------------------------\n",
+		s, "\n",
+		"----------------------------\n"}, ""))
+}
+
+// justifyLines writes s out to w, adding newlines between words
+// to keep line length under 'limit'. Newlines are indented
+// 'indent' spaces.
+func justifyLines(w *bufio.Writer, s string, limit int, indent int) {
+	n := 0
+	startOfLine := true
+	words := strings.Split(s, " ") // This only handles spaces (newlines, tabs ignored).
+	for _, x := range words {
+		if n+len(x) > limit && !startOfLine {
+			w.WriteString("\n")
+			w.WriteString(strings.Repeat(" ", indent))
+			n = indent
+			startOfLine = true
+		}
+		if startOfLine {
+			w.WriteString(x)
+			n += len(x)
+		} else {
+			w.WriteString(" " + x)
+			n += len(x) + 1
+		}
+		startOfLine = false
+	}
+}
+
+// pct prints a percentage representation of (total-bad)/total
+func pct(total, bad int64) string {
+	if bad == 0 || total == 0 {
+		return "100"
+	}
+	pct := 100.0 * float64(total-bad) / float64(total)
+	if pct > 99.9 {
+		return fmt.Sprintf("%2.5f", pct)
+	}
+	if pct > 95.0 {
+		return fmt.Sprintf("%2.3f", pct)
+	}
+	return fmt.Sprintf("%2.0f", pct)
+}
diff --git a/internal/reports/report_v2.go b/internal/reports/report_v2.go
new file mode 100644
index 0000000..381af21
--- /dev/null
+++ b/internal/reports/report_v2.go
@@ -0,0 +1,297 @@
+// Copyright 2023 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.
+
+// Implements structured report generation for Harbourbridge.
+package reports
+
+import (
+	"strings"
+	"time"
+
+	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
+)
+
+type Summary struct {
+	Text string `json:"text"`
+	Rating string `json:"rating"`
+	DbName string `json:"dbName"`
+}
+
+type IgnoredStatement struct {
+	StatementType string `json:"statementType"`
+	Statement     string `json:"statement"`
+}
+
+type ConversionMetadata struct {
+	ConversionType string        `json:"conversionType"`
+	Duration       time.Duration `json:"duration"`
+}
+
+type StatementStat struct {
+	Statement  string `json:"statement"`
+	Schema     int64  `json:"schema"`
+	Data       int64  `json:"data"`
+	Skip       int64  `json:"skip"`
+	Error      int64  `json:"error"`
+	TotalCount int64  `json:"totalCount"`
+}
+
+type StatementStats struct {
+	DriverName     string          `json:"driverName"`
+	StatementStats []StatementStat `json:"statementStats"`
+}
+
+type NameChange struct {
+	NameChangeType string `json:"nameChangeType"`
+	SourceTable    string `json:"sourceTable"`
+	OldName        string `json:"oldName"`
+	NewName        string `json:"newName"`
+}
+
+type Warnings struct {
+	WarningType string   `json:"warningType"`
+	WarningList []string `json:"warningList"`
+}
+
+type SchemaReport struct {
+	Rating       string `json:"rating"`
+	PkMissing    bool   `json:"pkMissing"`
+	Warnings     int64  `json:"warnings"`
+	TotalColumns int64  `json:"totalColumns"`
+}
+
+type DataReport struct {
+	Rating    string `json:"rating"`
+	BadRows   int64  `json:"badRows"`
+	TotalRows int64  `json:"totalRows"`
+	DryRun    bool   `json:"dryRun"`
+}
+
+type TableReport struct {
+	SrcTableName  string       `json:"srcTableName"`
+	SpTableName   string       `json:"spTableName"`
+	SchemaReport  SchemaReport `json:"schemaReport"`
+	DataReport    DataReport   `json:"dataReport"`
+	Warnings      []Warnings   `json:"warnings"`
+}
+
+type UnexpectedCondition struct {
+	Count     int64  `json:"count"`
+	Condition string `json:"condition"`
+}
+
+type UnexpectedConditions struct {
+	Reparsed             int64
+	UnexpectedConditions []UnexpectedCondition `json:"unexpectedConditions"`
+}
+
+type StructuredReport struct {
+	Summary              Summary              `json:"summary"`
+	IgnoredStatements    []IgnoredStatement   `json:"ignoredStatements"`
+	ConversionMetadata   []ConversionMetadata `json:"conversionMetadata"`
+	MigrationType        string               `json:"migrationType"`
+	StatementStats       StatementStats       `json:"statementStats"`
+	NameChanges          []NameChange         `json:"nameChanges"`
+	TableReports         []TableReport        `json:"tableReports"`
+	UnexpectedConditions UnexpectedConditions `json:"unexpectedConditions"`
+	SchemaOnly           bool                 `json:"-"`
+}
+
+// A report consists of the following parts:
+// 1. Summary (overall quality of conversion)
+// 2. Ignored statements
+// 3. Conversion duration
+// 4. Migration Type
+// 5. Statement stats (in case of dumps)
+// 6. Name changes
+// 7. Individual table reports (Detailed + Quality of conversion for each)
+// 8. Unexpected conditions
+//
+// This method the RAW structured report in JSON format. Several utilities can be built on top of
+// this raw, nested JSON data to output the reports in different user and machine friendly formats
+// such as CSV, TXT etc.
+func GenerateStructuredReport(driverName string, dbName string, conv *internal.Conv, badWrites map[string]int64, printTableReports bool, printUnexpecteds bool) StructuredReport {
+	//Create report object
+	var hbReport = StructuredReport{}
+	tableReports := AnalyzeTables(conv, badWrites)
+	//1. Generate summary
+	rating, summary := GenerateSummary(conv, tableReports, badWrites)
+	hbReport.Summary = Summary{Text: summary, Rating: rating, DbName: dbName}
+
+	//2. Ignored Statements
+	hbReport.IgnoredStatements = fetchIgnoredStatements(conv)
+
+	//3. Conversion Metadata
+	hbReport.ConversionMetadata = append(hbReport.ConversionMetadata, ConversionMetadata{ConversionType: "Schema", Duration: conv.Audit.SchemaConversionDuration})
+	hbReport.ConversionMetadata = append(hbReport.ConversionMetadata, ConversionMetadata{ConversionType: "Data", Duration: conv.Audit.DataConversionDuration})
+
+	//4. Migration Type
+	hbReport.MigrationType = mapMigrationType(*conv.Audit.MigrationType)
+
+	//5. Statement statistics
+	var isDump bool
+	if strings.Contains(driverName, "dump") {
+		isDump = true
+	}
+	if isDump {
+		hbReport.StatementStats.DriverName = driverName
+		hbReport.StatementStats.StatementStats = fetchStatementStats(driverName, conv)
+	}
+
+	//6. Name changes
+	hbReport.NameChanges = fetchNameChanges(conv)
+
+	//7. Table Reports
+	if printTableReports {
+		hbReport.TableReports = fetchTableReports(tableReports, conv)
+	}
+
+	//8. Unexpected Conditions
+	if printUnexpecteds {
+		hbReport.UnexpectedConditions = fetchUnexceptedConditions(driverName, conv)
+	}
+
+	return hbReport
+}
+
+func mapMigrationType(migrationType migration.MigrationData_MigrationType) string {
+	if migrationType == migration.MigrationData_DATA_ONLY {
+		return "DATA"
+	}
+	if migrationType == migration.MigrationData_SCHEMA_AND_DATA {
+		return "SCHEMA_AND_DATA"
+	}
+	if migrationType == migration.MigrationData_SCHEMA_ONLY {
+		return "SCHEMA"
+	}
+	return "UNSPECIFIED"
+}
+
+func fetchIgnoredStatements(conv *internal.Conv) (ignoredStatements []IgnoredStatement) {
+	for s := range conv.Stats.Statement {
+		switch s {
+		case "CreateFunctionStmt":
+			ignoredStatements = append(ignoredStatements, IgnoredStatement{StatementType: "function", Statement: s})
+		case "CreateSeqStmt", "CreateSequenceStmt":
+			ignoredStatements = append(ignoredStatements, IgnoredStatement{StatementType: "sequence", Statement: s})
+		case "CreatePLangStmt", "CreateProcedureStmt":
+			ignoredStatements = append(ignoredStatements, IgnoredStatement{StatementType: "procedure", Statement: s})
+		case "CreateTrigStmt":
+			ignoredStatements = append(ignoredStatements, IgnoredStatement{StatementType: "trigger", Statement: s})
+		case "IndexStmt", "CreateIndexStmt":
+			ignoredStatements = append(ignoredStatements, IgnoredStatement{StatementType: "(non-primary) index", Statement: s})
+		case "ViewStmt", "CreateViewStmt":
+			ignoredStatements = append(ignoredStatements, IgnoredStatement{StatementType: "view", Statement: s})
+		}
+	}
+	return ignoredStatements
+}
+
+func fetchStatementStats(driverName string, conv *internal.Conv) (statementStats []StatementStat) {
+	for s, x := range conv.Stats.Statement {
+		statementStats = append(statementStats, StatementStat{Statement: s, Schema: x.Schema, Data: x.Data, Skip: x.Skip, Error: x.Error})
+	}
+	return statementStats
+}
+
+func fetchNameChanges(conv *internal.Conv) (nameChanges []NameChange) {
+	for tableId, spTable := range conv.SpSchema {
+		srcTable := conv.SrcSchema[tableId]
+		if srcTable.Name != spTable.Name {
+			nameChanges = append(nameChanges, NameChange{NameChangeType: "TableName", SourceTable: srcTable.Name, OldName: srcTable.Name, NewName: spTable.Name})
+		}
+		for colId, spCol := range spTable.ColDefs {
+			srcCol, ok := srcTable.ColDefs[colId]
+			if !ok {
+				continue
+			}
+			if srcCol.Name != spCol.Name {
+				nameChanges = append(nameChanges, NameChange{NameChangeType: "ColumnName", SourceTable: srcTable.Name, OldName: srcCol.Name, NewName: spCol.Name})
+			}
+		}
+		for _, spFk := range conv.SpSchema[tableId].ForeignKeys {
+			srcFk, err := internal.GetSrcFkFromId(conv.SrcSchema[tableId].ForeignKeys, spFk.Id)
+			if err != nil {
+				continue
+			}
+			if srcFk.Name != spFk.Name {
+				nameChanges = append(nameChanges, NameChange{NameChangeType: "ForeignKey", SourceTable: srcTable.Name, OldName: srcFk.Name, NewName: spFk.Name})
+			}
+		}
+		for _, spIdx := range conv.SpSchema[tableId].Indexes {
+			srcIdx, err := internal.GetSrcIndexFromId(conv.SrcSchema[tableId].Indexes, spIdx.Id)
+			if err != nil {
+				continue
+			}
+			if srcIdx.Name != spIdx.Name {
+				nameChanges = append(nameChanges, NameChange{NameChangeType: "Index", SourceTable: srcTable.Name, OldName: srcIdx.Name, NewName: spIdx.Name})
+			}
+		}
+	}
+	return nameChanges
+}
+
+func fetchTableReports(inputTableReports []tableReport, conv *internal.Conv) (tableReports []TableReport) {
+	for _, t := range inputTableReports {
+		//1. src and Sp Table Names
+		tableReport := TableReport{SrcTableName: conv.SrcSchema[t.SrcTable].Name}
+		tableReport.SpTableName = conv.SrcSchema[t.SrcTable].Name
+
+		//2. Schema Report
+		migrationType := *conv.Audit.MigrationType
+		if migrationType != migration.MigrationData_DATA_ONLY {
+			tableReport.SchemaReport = getSchemaReport(t.Cols, t.Warnings, t.SyntheticPKey != "")
+		}
+		//3. Data Report
+		schemaOnly := conv.SchemaMode()
+		if !schemaOnly {
+			tableReport.DataReport = getDataReport(t.rows, t.badRows, conv.Audit.DryRun)
+		}
+		//4. Warnings
+		for _, x := range t.Body {
+			var warnings = Warnings{WarningType: x.Heading}
+			for _, l := range x.Lines {
+				warnings.WarningList = append(warnings.WarningList, l)
+			}
+			tableReport.Warnings = append(tableReport.Warnings, warnings)
+		}
+		tableReports = append(tableReports, tableReport)
+	}
+	return tableReports
+}
+
+func getSchemaReport(cols, warnings int64, missingPKey bool) (schemaReport SchemaReport) {
+	schemaReport.TotalColumns = cols
+	schemaReport.Warnings = warnings
+	schemaReport.PkMissing = missingPKey
+	schemaReport.Rating, _ = RateSchema(cols, warnings, missingPKey, false)
+	return schemaReport
+}
+
+func getDataReport(rows int64, badRows int64, dryRun bool) (dataReport DataReport) {
+	dataReport.DryRun = dryRun
+	dataReport.TotalRows = rows
+	dataReport.BadRows = badRows
+	dataReport.Rating, _ = rateData(rows, badRows, dryRun)
+	return dataReport
+}
+
+func fetchUnexceptedConditions(driverName string, conv *internal.Conv) (unexpectedConditions UnexpectedConditions) {
+	unexpectedConditions.Reparsed = conv.Stats.Reparsed
+	for s, n := range conv.Stats.Unexpected {
+		unexpectedConditions.UnexpectedConditions = append(unexpectedConditions.UnexpectedConditions, UnexpectedCondition{Count: n, Condition: s})
+	}
+	return unexpectedConditions
+}
diff --git a/webv2/summary/summary_service.go b/webv2/summary/summary_service.go
index 97e19b4..3552616 100644
--- a/webv2/summary/summary_service.go
+++ b/webv2/summary/summary_service.go
@@ -1,14 +1,14 @@
 package summary
 
 import (
-	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/internal/reports"
 	"github.com/cloudspannerecosystem/harbourbridge/webv2/session"
 )
 
 // getSummary returns table wise summary of conversion.
 func getSummary() map[string]ConversionSummary {
 	sessionState := session.GetSessionState()
-	reports := internal.AnalyzeTables(sessionState.Conv, nil)
+	reports := reports.AnalyzeTables(sessionState.Conv, nil)
 
 	summary := make(map[string]ConversionSummary)
 	for _, t := range reports {
diff --git a/webv2/utilities/utilities.go b/webv2/utilities/utilities.go
index ea574c2..37132f0 100644
--- a/webv2/utilities/utilities.go
+++ b/webv2/utilities/utilities.go
@@ -309,7 +309,7 @@ func GetFilePrefix(now time.Time) (string, error) {
 			return "", fmt.Errorf("Can not create database name : %v", err)
 		}
 	}
-	return dbName + ".", nil
+	return dbName, nil
 }
 
 func UpdateDataType(conv *internal.Conv, newType, tableId, colId string) error {
diff --git a/webv2/web.go b/webv2/web.go
index 79ca0ea..a64f9fc 100644
--- a/webv2/web.go
+++ b/webv2/web.go
@@ -39,6 +39,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/internal/reports"
 	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
@@ -871,10 +872,10 @@ func addIndex(newIndex ddl.CreateIndex) (ddl.CreateIndex, error) {
 // getConversionRate returns table wise color coded conversion rate.
 func getConversionRate(w http.ResponseWriter, r *http.Request) {
 	sessionState := session.GetSessionState()
-	reports := internal.AnalyzeTables(sessionState.Conv, nil)
+	hb_reports := reports.AnalyzeTables(sessionState.Conv, nil)
 	rate := make(map[string]string)
-	for _, t := range reports {
-		rate[t.SpTable], _ = internal.RateSchema(t.Cols, t.Warnings, t.SyntheticPKey != "", false)
+	for _, t := range hb_reports {
+		rate[t.SpTable], _ = reports.RateSchema(t.Cols, t.Warnings, t.SyntheticPKey != "", false)
 	}
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(rate)
@@ -910,9 +911,9 @@ func getReportFile(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Can not get file prefix : %v", err), http.StatusInternalServerError)
 	}
-	reportFileName := "frontend/" + filePrefix + "report.txt"
+	reportFileName := "frontend/" + filePrefix
 	sessionState := session.GetSessionState()
-	conversion.Report(sessionState.Driver, nil, ioHelper.BytesRead, "", sessionState.Conv, reportFileName, ioHelper.Out)
+	conversion.Report(sessionState.Driver, nil, ioHelper.BytesRead, "", sessionState.Conv, reportFileName, sessionState.DbName, ioHelper.Out)
 	reportAbsPath, err := filepath.Abs(reportFileName)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Can not create absolute path : %v", err), http.StatusInternalServerError)
@@ -2193,7 +2194,7 @@ func addTypeToList(convertedType string, spType string, issues []internal.Schema
 		if len(issues) > 0 {
 			var briefs []string
 			for _, issue := range issues {
-				briefs = append(briefs, internal.IssueDB[issue].Brief)
+				briefs = append(briefs, reports.IssueDB[issue].Brief)
 			}
 			l = append(l, typeIssue{T: spType, Brief: fmt.Sprintf(strings.Join(briefs, ", "))})
 		} else {


Test patch:
diff --git a/sources/mysql/report_test.go b/sources/mysql/report_test.go
index e014436..f910a51 100644
--- a/sources/mysql/report_test.go
+++ b/sources/mysql/report_test.go
@@ -17,12 +17,15 @@ package mysql
 import (
 	"bufio"
 	"bytes"
-	"fmt"
+	"encoding/json"
+	"io/ioutil"
+	"path/filepath"
 	"strings"
 	"testing"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/internal/reports"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/stretchr/testify/assert"
@@ -72,110 +75,15 @@ func TestReport(t *testing.T) {
 	}
 	buf := new(bytes.Buffer)
 	w := bufio.NewWriter(buf)
-	internal.GenerateReport(constants.MYSQLDUMP, conv, w, badWrites, true, true)
+	actualStructuredReport := reports.GenerateStructuredReport(constants.MYSQLDUMP, "sampleDB", conv, badWrites, true, true)
+	var expectedStructuredReport reports.StructuredReport
+	expectedBytes, _ := ioutil.ReadFile(filepath.Join("..", "..", "test_data", "mysql_structured_report.json"))
+	_ = json.Unmarshal(expectedBytes, &expectedStructuredReport)
+	assert.Equal(t, expectedStructuredReport, actualStructuredReport)
+	reports.GenerateTextReport(actualStructuredReport, w)
 	w.Flush()
-	// Print copy of report to stdout (shows up when running go test -v).
-	fmt.Print(buf.String())
-	// Do a dumb comparison with a static 'expected' string.
-	// If 'expected' is painful to maintain, delete it and just
-	// use this test as a smoke test and a way to see report output
-	// for some canned/sample data.
-	expected :=
-		`----------------------------
-Summary of Conversion
-----------------------------
-Schema conversion: POOR (65% of 17006 columns mapped cleanly) + some missing primary keys.
-Data conversion: POOR (66% of 6000 rows written to Spanner).
-
-The remainder of this report provides stats on the mysqldump statements
-processed, followed by 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.
-
-----------------------------
-Statements Processed
-----------------------------
-Analysis of statements in mysqldump output, broken down by statement type.
-  schema: statements successfully processed for Spanner schema information.
-    data: statements successfully processed for data.
-    skip: statements not relevant for Spanner schema or data.
-   error: statements that could not be processed.
-  --------------------------------------
-  schema   data   skip  error  statement
-  --------------------------------------
-       5      0      0      0  CreateTableStmt
-See https://github.com/pingcap/parser for definitions of statement types
-(pingcap/tidb/parser is the library we use for parsing mysqldump output).
-
------------------------------------------------------------------------------------------------------
-Name Changes in Migration
------------------------------------------------------------------------------------------------------
-             Source Table          Change                  Old Name                  New Name
------------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------------
-
-
-----------------------------
-Table bad_schema
-----------------------------
-Schema conversion: POOR (50% of 2 columns mapped cleanly) + missing primary key.
-Data conversion: OK (94% of 1000 rows written to Spanner).
-
-Warnings
-1) Column 'synth_id' was added because this table didn't have a primary key.
-   Spanner requires a primary key for every table.
-2) Some columns will consume more storage in Spanner e.g. for column 'a', source
-   DB type float is mapped to Spanner data type float64.
-
-----------------------------
-Table default_value
-----------------------------
-Schema conversion: POOR (50% of 2 columns mapped cleanly).
-Data conversion: NONE (no data rows found).
-
-Warning
-1) Some columns have default values which Spanner does not support e.g. column
-   'b'.
-
-----------------------------
-Table excellent_schema
-----------------------------
-Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
-Data conversion: NONE (no data rows found).
-
-----------------------------
-Table foreign_key
-----------------------------
-Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
-Data conversion: NONE (no data rows found).
-
-----------------------------
-Table no_pk
-----------------------------
-Schema conversion: POOR (67% of 3 columns mapped cleanly) + missing primary key.
-Data conversion: POOR (60% of 5000 rows written to Spanner).
-
-Warnings
-1) Column 'synth_id' was added because this table didn't have a primary key.
-   Spanner requires a primary key for every table.
-2) Some columns will consume more storage in Spanner e.g. for column 'b', source
-   DB type int(11) is mapped to Spanner data type int64.
-
-----------------------------
-Unexpected Conditions
-----------------------------
-For debugging only. This section provides details of unexpected conditions
-encountered as we processed the mysqldump data. In particular, the AST node
-representation used by the pingcap/tidb/parser library used for parsing
-mysqldump output is highly permissive: almost any construct can appear at
-any node in the AST tree. The list details all unexpected nodes and
-conditions.
-  --------------------------------------
-   count  condition
-  --------------------------------------
-       5  Testing unexpected messages
-
-`
-	assert.Equal(t, expected, buf.String())
+	expectedBytes, _ = ioutil.ReadFile(filepath.Join("..", "..", "test_data", "mysql_text_report.txt"))
+	expected := string(expectedBytes)
+	actual := buf.String()
+	assert.Equal(t, expected, actual)
 }
diff --git a/sources/postgres/report_test.go b/sources/postgres/report_test.go
index 192de92..888d699 100644
--- a/sources/postgres/report_test.go
+++ b/sources/postgres/report_test.go
@@ -17,12 +17,15 @@ package postgres
 import (
 	"bufio"
 	"bytes"
-	"fmt"
+	"encoding/json"
+	"io/ioutil"
+	"path/filepath"
 	"strings"
 	"testing"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/internal/reports"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/stretchr/testify/assert"
@@ -68,113 +71,15 @@ func TestReport(t *testing.T) {
 	}
 	buf := new(bytes.Buffer)
 	w := bufio.NewWriter(buf)
-	internal.GenerateReport(constants.PGDUMP, conv, w, badWrites, true, true)
+	actualStructuredReport := reports.GenerateStructuredReport(constants.PGDUMP, "sampleDB", conv, badWrites, true, true)
+	var expectedStructuredReport reports.StructuredReport
+	expectedBytes, _ := ioutil.ReadFile(filepath.Join("..", "..", "test_data", "postgres_structured_report.json"))
+	_ = json.Unmarshal(expectedBytes, &expectedStructuredReport)
+	assert.Equal(t, expectedStructuredReport, actualStructuredReport)
+	reports.GenerateTextReport(actualStructuredReport, w)
 	w.Flush()
-	// Print copy of report to stdout (shows up when running go test -v).
-	fmt.Print(buf.String())
-	// Do a dumb comparison with a static 'expected' string.
-	// If 'expected' is painful to maintain, delete it and just
-	// use this test as a smoke test and a way to see report output
-	// for some canned/sample data.
-	expected :=
-		`----------------------------
-Summary of Conversion
-----------------------------
-Schema conversion: POOR (58% of 19006 columns mapped cleanly) + some missing primary keys.
-Data conversion: POOR (66% of 6000 rows written to Spanner).
-
-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. 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.
-
-----------------------------
-Statements Processed
-----------------------------
-Analysis of statements in pg_dump output, broken down by statement type.
-  schema: statements successfully processed for Spanner schema information.
-    data: statements successfully processed for data.
-    skip: statements not relevant for Spanner schema or data.
-   error: statements that could not be processed.
-  --------------------------------------
-  schema   data   skip  error  statement
-  --------------------------------------
-       5      0      0      0  CreateStmt
-See github.com/pganalyze/pg_query_go for definitions of statement types
-(pganalyze/pg_query_go is the library we use for parsing pg_dump output).
-
------------------------------------------------------------------------------------------------------
-Name Changes in Migration
------------------------------------------------------------------------------------------------------
-             Source Table          Change                  Old Name                  New Name
------------------------------------------------------------------------------------------------------
------------------------------------------------------------------------------------------------------
-
-
-----------------------------
-Table bad_schema
-----------------------------
-Schema conversion: POOR (25% of 4 columns mapped cleanly) + missing primary key.
-Data conversion: OK (94% of 1000 rows written to Spanner).
-
-Warnings
-1) Column 'synth_id' was added because this table didn't have a primary key.
-   Spanner requires a primary key for every table.
-2) Some columns will consume more storage in Spanner e.g. for column 'b', source
-   DB type int4 is mapped to Spanner data type int64.
-3) Column 'c': type int4[4][2] is mapped to string(max). Spanner doesn't support
-   multi-dimensional arrays.
-4) Column 'd': type circle is mapped to string(max). No appropriate Spanner
-   type.
-
-----------------------------
-Table default_value
-----------------------------
-Schema conversion: POOR (50% of 2 columns mapped cleanly).
-Data conversion: NONE (no data rows found).
-
-Warning
-1) Some columns have default values which Spanner does not support e.g. column
-   'b'.
-
-----------------------------
-Table excellent_schema
-----------------------------
-Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
-Data conversion: NONE (no data rows found).
-
-----------------------------
-Table foreign_key
-----------------------------
-Schema conversion: EXCELLENT (all 2 columns mapped cleanly).
-Data conversion: NONE (no data rows found).
-
-----------------------------
-Table no_pk
-----------------------------
-Schema conversion: POOR (67% of 3 columns mapped cleanly) + missing primary key.
-Data conversion: POOR (60% of 5000 rows written to Spanner).
-
-Warnings
-1) Column 'synth_id' was added because this table didn't have a primary key.
-   Spanner requires a primary key for every table.
-2) Some columns will consume more storage in Spanner e.g. for column 'b', source
-   DB type int4 is mapped to Spanner data type int64.
-
-----------------------------
-Unexpected Conditions
-----------------------------
-For debugging only. This section provides details of unexpected conditions
-encountered as we processed the pg_dump data. In particular, the AST node
-representation used by the pganalyze/pg_query_go library used for parsing
-pg_dump output is highly permissive: almost any construct can appear at
-any node in the AST tree. The list details all unexpected nodes and
-conditions.
-  --------------------------------------
-   count  condition
-  --------------------------------------
-       5  Testing unexpected messages
-
-`
-	assert.Equal(t, expected, expected)
+	expectedBytes, _ = ioutil.ReadFile(filepath.Join("..", "..", "test_data", "postgres_text_report.txt"))
+	expected := string(expectedBytes)
+	actual := buf.String()
+	assert.Equal(t, expected, actual)
 }
diff --git a/test_data/mysql_structured_report.json b/test_data/mysql_structured_report.json
new file mode 100644
index 0000000..5831625
--- /dev/null
+++ b/test_data/mysql_structured_report.json
@@ -0,0 +1,152 @@
+{
+   "summary":{
+      "text":"Schema conversion: POOR (65% of 17006 columns mapped cleanly) + some missing primary keys.\nData conversion: POOR (66% of 6000 rows written to Spanner).\n",
+      "rating":"POOR",
+      "dbName": "sampleDB"
+   },
+   "ignoredStatements":null,
+   "conversionMetadata":[
+      {
+         "conversionType":"Schema",
+         "duration":0
+      },
+      {
+         "conversionType":"Data",
+         "duration":0
+      }
+   ],
+   "migrationType":"SCHEMA_AND_DATA",
+   "statementStats":{
+      "driverName":"mysqldump",
+      "statementStats":[
+         {
+            "statement":"CreateTableStmt",
+            "schema":5,
+            "data":0,
+            "skip":0,
+            "error":0,
+            "totalCount":0
+         }
+      ]
+   },
+   "nameChanges":null,
+   "tableReports":[
+      {
+         "srcTableName":"bad_schema",
+         "spTableName":"bad_schema",
+         "schemaReport":{
+            "rating":"POOR",
+            "pkMissing":true,
+            "warnings":1,
+            "totalColumns":2
+         },
+         "dataReport":{
+            "rating":"OK",
+            "badRows":60,
+            "totalRows":1000,
+            "dryRun":false
+         },
+         "warnings":[
+            {
+               "warningType":"Warnings",
+               "warningList":[
+                  "Column 'synth_id' was added because this table didn't have a primary key. Spanner requires a primary key for every table",
+                  "Some columns will consume more storage in Spanner e.g. for column 'a', source DB type float is mapped to Spanner data type float64"
+               ]
+            }
+         ]
+      },
+      {
+         "srcTableName":"default_value",
+         "spTableName":"default_value",
+         "schemaReport":{
+            "rating":"POOR",
+            "pkMissing":false,
+            "warnings":1,
+            "totalColumns":2
+         },
+         "dataReport":{
+            "rating":"NONE",
+            "badRows":0,
+            "totalRows":0,
+            "dryRun":false
+         },
+         "warnings":[
+            {
+               "warningType":"Warning",
+               "warningList":[
+                  "Some columns have default values which Spanner does not support e.g. column 'b'"
+               ]
+            }
+         ]
+      },
+      {
+         "srcTableName":"excellent_schema",
+         "spTableName":"excellent_schema",
+         "schemaReport":{
+            "rating":"EXCELLENT",
+            "pkMissing":false,
+            "warnings":0,
+            "totalColumns":2
+         },
+         "dataReport":{
+            "rating":"NONE",
+            "badRows":0,
+            "totalRows":0,
+            "dryRun":false
+         },
+         "warnings":null
+      },
+      {
+         "srcTableName":"foreign_key",
+         "spTableName":"foreign_key",
+         "schemaReport":{
+            "rating":"EXCELLENT",
+            "pkMissing":false,
+            "warnings":0,
+            "totalColumns":2
+         },
+         "dataReport":{
+            "rating":"NONE",
+            "badRows":0,
+            "totalRows":0,
+            "dryRun":false
+         },
+         "warnings":null
+      },
+      {
+         "srcTableName":"no_pk",
+         "spTableName":"no_pk",
+         "schemaReport":{
+            "rating":"POOR",
+            "pkMissing":true,
+            "warnings":1,
+            "totalColumns":3
+         },
+         "dataReport":{
+            "rating":"POOR",
+            "badRows":2000,
+            "totalRows":5000,
+            "dryRun":false
+         },
+         "warnings":[
+            {
+               "warningType":"Warnings",
+               "warningList":[
+                  "Column 'synth_id' was added because this table didn't have a primary key. Spanner requires a primary key for every table",
+                  "Some columns will consume more storage in Spanner e.g. for column 'b', source DB type int(11) is mapped to Spanner data type int64"
+               ]
+            }
+         ]
+      }
+   ],
+   "unexpectedConditions":{
+      "Reparsed":0,
+      "unexpectedConditions":[
+         {
+            "count":5,
+            "condition":"Testing unexpected messages"
+         }
+      ]
+   }
+}
\ No newline at end of file
diff --git a/test_data/mysql_text_report.txt b/test_data/mysql_text_report.txt
new file mode 100644
index 0000000..fb2e03a
--- /dev/null
+++ b/test_data/mysql_text_report.txt
@@ -0,0 +1,88 @@
+----------------------------
+Summary of Conversion
+----------------------------
+Schema conversion: POOR (65% of 17006 columns mapped cleanly) + some missing primary keys.
+Data conversion: POOR (66% of 6000 rows written to Spanner).
+
+The remainder of this report provides stats on the mysqldump statements
+processed, followed by 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.
+
+----------------------------
+Statements Processed
+----------------------------
+Analysis of statements in mysqldump output, broken down by statement type.
+  schema: statements successfully processed for Spanner schema information.
+    data: statements successfully processed for data.
+    skip: statements not relevant for Spanner schema or data.
+   error: statements that could not be processed.
+  --------------------------------------
+  schema   data   skip  error  statement
+  --------------------------------------
+       5      0      0      0  CreateTableStmt
+See https://github.com/pingcap/parser for definitions of statement types
+(pingcap/tidb/parser is the library we use for parsing mysqldump output).
+
+No Name Changes in Migration
+----------------------------
+Table bad_schema
+----------------------------
+Schema conversion: POOR (50% of 2 columns mapped cleanly) + missing primary key.
+Data conversion: OK (94% of 1000 rows written to Spanner).
+
+Warnings
+1) Column 'synth_id' was added because this table didn't have a primary key.
+   Spanner requires a primary key for every table.
+2) Some columns will consume more storage in Spanner e.g. for column 'a', source
+   DB type float is mapped to Spanner data type float64.
+
+----------------------------
+Table default_value
+----------------------------
+Schema conversion: POOR (50% of 2 columns mapped cleanly).
+Data conversion: NONE (100% of 0 rows written to Spanner).
+
+Warning
+1) Some columns have default values which Spanner does not support e.g. column
+   'b'.
+
+----------------------------
+Table excellent_schema
+----------------------------
+Schema conversion: EXCELLENT (100% of 2 columns mapped cleanly).
+Data conversion: NONE (100% of 0 rows written to Spanner).
+
+----------------------------
+Table foreign_key
+----------------------------
+Schema conversion: EXCELLENT (100% of 2 columns mapped cleanly).
+Data conversion: NONE (100% of 0 rows written to Spanner).
+
+----------------------------
+Table no_pk
+----------------------------
+Schema conversion: POOR (67% of 3 columns mapped cleanly) + missing primary key.
+Data conversion: POOR (60% of 5000 rows written to Spanner).
+
+Warnings
+1) Column 'synth_id' was added because this table didn't have a primary key.
+   Spanner requires a primary key for every table.
+2) Some columns will consume more storage in Spanner e.g. for column 'b', source
+   DB type int(11) is mapped to Spanner data type int64.
+
+----------------------------
+Unexpected Conditions
+----------------------------
+For debugging only. This section provides details of unexpected conditions
+encountered as we processed the mysqldump data. In particular, the AST node
+representation used by the pingcap/tidb/parser library used for parsing
+mysqldump output is highly permissive: almost any construct can appear at
+any node in the AST tree. The list details all unexpected nodes and
+conditions.
+  --------------------------------------
+   count  condition
+  --------------------------------------
+       5  Testing unexpected messages
+
diff --git a/test_data/postgres_structured_report.json b/test_data/postgres_structured_report.json
new file mode 100644
index 0000000..87c33aa
--- /dev/null
+++ b/test_data/postgres_structured_report.json
@@ -0,0 +1,154 @@
+{
+    "summary":{
+       "text":"Schema conversion: POOR (58% of 19006 columns mapped cleanly) + some missing primary keys.\nData conversion: POOR (66% of 6000 rows written to Spanner).\n",
+       "rating":"POOR",
+       "dbName": "sampleDB"
+    },
+    "ignoredStatements":null,
+    "conversionMetadata":[
+       {
+          "conversionType":"Schema",
+          "duration":0
+       },
+       {
+          "conversionType":"Data",
+          "duration":0
+       }
+    ],
+    "migrationType":"SCHEMA_AND_DATA",
+    "statementStats":{
+       "driverName":"pg_dump",
+       "statementStats":[
+          {
+             "statement":"CreateStmt",
+             "schema":5,
+             "data":0,
+             "skip":0,
+             "error":0,
+             "totalCount":0
+          }
+       ]
+    },
+    "nameChanges":null,
+    "tableReports":[
+       {
+          "srcTableName":"bad_schema",
+          "spTableName":"bad_schema",
+          "schemaReport":{
+             "rating":"POOR",
+             "pkMissing":true,
+             "warnings":3,
+             "totalColumns":4
+          },
+          "dataReport":{
+             "rating":"OK",
+             "badRows":60,
+             "totalRows":1000,
+             "dryRun":false
+          },
+          "warnings":[
+             {
+                "warningType":"Warnings",
+                "warningList":[
+                   "Column 'synth_id' was added because this table didn't have a primary key. Spanner requires a primary key for every table",
+                   "Some columns will consume more storage in Spanner e.g. for column 'b', source DB type int4 is mapped to Spanner data type int64",
+                   "Column 'c': type int4[4][2] is mapped to string(max). Spanner doesn't support multi-dimensional arrays",
+                   "Column 'd': type circle is mapped to string(max). No appropriate Spanner type"
+                ]
+             }
+          ]
+       },
+       {
+          "srcTableName":"default_value",
+          "spTableName":"default_value",
+          "schemaReport":{
+             "rating":"POOR",
+             "pkMissing":false,
+             "warnings":1,
+             "totalColumns":2
+          },
+          "dataReport":{
+             "rating":"NONE",
+             "badRows":0,
+             "totalRows":0,
+             "dryRun":false
+          },
+          "warnings":[
+             {
+                "warningType":"Warning",
+                "warningList":[
+                   "Some columns have default values which Spanner does not support e.g. column 'b'"
+                ]
+             }
+          ]
+       },
+       {
+          "srcTableName":"excellent_schema",
+          "spTableName":"excellent_schema",
+          "schemaReport":{
+             "rating":"EXCELLENT",
+             "pkMissing":false,
+             "warnings":0,
+             "totalColumns":2
+          },
+          "dataReport":{
+             "rating":"NONE",
+             "badRows":0,
+             "totalRows":0,
+             "dryRun":false
+          },
+          "warnings":null
+       },
+       {
+          "srcTableName":"foreign_key",
+          "spTableName":"foreign_key",
+          "schemaReport":{
+             "rating":"EXCELLENT",
+             "pkMissing":false,
+             "warnings":0,
+             "totalColumns":2
+          },
+          "dataReport":{
+             "rating":"NONE",
+             "badRows":0,
+             "totalRows":0,
+             "dryRun":false
+          },
+          "warnings":null
+       },
+       {
+          "srcTableName":"no_pk",
+          "spTableName":"no_pk",
+          "schemaReport":{
+             "rating":"POOR",
+             "pkMissing":true,
+             "warnings":1,
+             "totalColumns":3
+          },
+          "dataReport":{
+             "rating":"POOR",
+             "badRows":2000,
+             "totalRows":5000,
+             "dryRun":false
+          },
+          "warnings":[
+             {
+                "warningType":"Warnings",
+                "warningList":[
+                   "Column 'synth_id' was added because this table didn't have a primary key. Spanner requires a primary key for every table",
+                   "Some columns will consume more storage in Spanner e.g. for column 'b', source DB type int4 is mapped to Spanner data type int64"
+                ]
+             }
+          ]
+       }
+    ],
+    "unexpectedConditions":{
+       "Reparsed":0,
+       "unexpectedConditions":[
+          {
+             "count":5,
+             "condition":"Testing unexpected messages"
+          }
+       ]
+    }
+ }
\ No newline at end of file
diff --git a/test_data/postgres_text_report.txt b/test_data/postgres_text_report.txt
new file mode 100644
index 0000000..5cfc404
--- /dev/null
+++ b/test_data/postgres_text_report.txt
@@ -0,0 +1,91 @@
+----------------------------
+Summary of Conversion
+----------------------------
+Schema conversion: POOR (58% of 19006 columns mapped cleanly) + some missing primary keys.
+Data conversion: POOR (66% of 6000 rows written to Spanner).
+
+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. 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.
+
+----------------------------
+Statements Processed
+----------------------------
+Analysis of statements in pg_dump output, broken down by statement type.
+  schema: statements successfully processed for Spanner schema information.
+    data: statements successfully processed for data.
+    skip: statements not relevant for Spanner schema or data.
+   error: statements that could not be processed.
+  --------------------------------------
+  schema   data   skip  error  statement
+  --------------------------------------
+       5      0      0      0  CreateStmt
+See github.com/pganalyze/pg_query_go for definitions of statement types
+(pganalyze/pg_query_go is the library we use for parsing pg_dump output).
+
+No Name Changes in Migration
+----------------------------
+Table bad_schema
+----------------------------
+Schema conversion: POOR (25% of 4 columns mapped cleanly) + missing primary key.
+Data conversion: OK (94% of 1000 rows written to Spanner).
+
+Warnings
+1) Column 'synth_id' was added because this table didn't have a primary key.
+   Spanner requires a primary key for every table.
+2) Some columns will consume more storage in Spanner e.g. for column 'b', source
+   DB type int4 is mapped to Spanner data type int64.
+3) Column 'c': type int4[4][2] is mapped to string(max). Spanner doesn't support
+   multi-dimensional arrays.
+4) Column 'd': type circle is mapped to string(max). No appropriate Spanner
+   type.
+
+----------------------------
+Table default_value
+----------------------------
+Schema conversion: POOR (50% of 2 columns mapped cleanly).
+Data conversion: NONE (100% of 0 rows written to Spanner).
+
+Warning
+1) Some columns have default values which Spanner does not support e.g. column
+   'b'.
+
+----------------------------
+Table excellent_schema
+----------------------------
+Schema conversion: EXCELLENT (100% of 2 columns mapped cleanly).
+Data conversion: NONE (100% of 0 rows written to Spanner).
+
+----------------------------
+Table foreign_key
+----------------------------
+Schema conversion: EXCELLENT (100% of 2 columns mapped cleanly).
+Data conversion: NONE (100% of 0 rows written to Spanner).
+
+----------------------------
+Table no_pk
+----------------------------
+Schema conversion: POOR (67% of 3 columns mapped cleanly) + missing primary key.
+Data conversion: POOR (60% of 5000 rows written to Spanner).
+
+Warnings
+1) Column 'synth_id' was added because this table didn't have a primary key.
+   Spanner requires a primary key for every table.
+2) Some columns will consume more storage in Spanner e.g. for column 'b', source
+   DB type int4 is mapped to Spanner data type int64.
+
+----------------------------
+Unexpected Conditions
+----------------------------
+For debugging only. This section provides details of unexpected conditions
+encountered as we processed the pg_dump data. In particular, the AST node
+representation used by the pganalyze/pg_query_go library used for parsing
+pg_dump output is highly permissive: almost any construct can appear at
+any node in the AST tree. The list details all unexpected nodes and
+conditions.
+  --------------------------------------
+   count  condition
+  --------------------------------------
+       5  Testing unexpected messages
+
diff --git a/testing/dynamodb/snapshot/integration_test.go b/testing/dynamodb/snapshot/integration_test.go
index ae47cab..5ff1f03 100644
--- a/testing/dynamodb/snapshot/integration_test.go
+++ b/testing/dynamodb/snapshot/integration_test.go
@@ -215,7 +215,7 @@ func TestIntegration_DYNAMODB_Command(t *testing.T) {
 	now := time.Now()
 	dbName, _ := utils.GetDatabaseName(constants.DYNAMODB, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	filePrefix := filepath.Join(tmpdir, dbName+".")
+	filePrefix := filepath.Join(tmpdir, dbName)
 
 	args := fmt.Sprintf(`schema-and-data -source=%s -prefix=%s -target-profile="instance=%s,dbName=%s"`, constants.DYNAMODB, filePrefix, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
diff --git a/testing/dynamodb/streaming/integration_test.go b/testing/dynamodb/streaming/integration_test.go
index 267da2f..3f44d20 100644
--- a/testing/dynamodb/streaming/integration_test.go
+++ b/testing/dynamodb/streaming/integration_test.go
@@ -325,7 +325,7 @@ func TestIntegration_DYNAMODB_Streaming_Command(t *testing.T) {
 	now := time.Now()
 	dbName, _ := utils.GetDatabaseName(constants.DYNAMODB, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	filePrefix := filepath.Join(tmpdir, dbName+".")
+	filePrefix := filepath.Join(tmpdir, dbName)
 
 	client, err := spanner.NewClient(ctx, dbURI)
 	if err != nil {
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index ecd8c79..67299c4 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -119,7 +119,7 @@ func TestIntegration_MYSQL_SchemaAndDataSubcommand(t *testing.T) {
 
 	dbName := "mysql-dc-schema-and-data"
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	filePrefix := filepath.Join(tmpdir, dbName+".")
+	filePrefix := filepath.Join(tmpdir, dbName)
 
 	host, user, srcDb, password := os.Getenv("MYSQLHOST"), os.Getenv("MYSQLUSER"), os.Getenv("MYSQLDATABASE"), os.Getenv("MYSQLPWD")
 	envVars := common.ClearEnvVariables([]string{"MYSQLHOST", "MYSQLUSER", "MYSQLDATABASE", "MYSQLPWD"})
@@ -170,16 +170,16 @@ func TestIntegration_MySQLDUMP_SchemaSubcommand(t *testing.T) {
 	defer dropDatabase(t, dbURI)
 
 	dumpFilePath := "../../test_data/mysqldump.test.out"
-	filePrefix := filepath.Join(tmpdir, dbName+".")
-	sessionFile := fmt.Sprintf("%ssession.json", filePrefix)
+	filePrefix := filepath.Join(tmpdir, dbName)
+	sessionFile := fmt.Sprintf("%s.session.json", filePrefix)
 	runSchemaSubcommand(t, dbName, filePrefix, sessionFile, dumpFilePath)
-	if _, err := os.Stat(fmt.Sprintf("%sreport.txt", filePrefix)); os.IsNotExist(err) {
+	if _, err := os.Stat(fmt.Sprintf("%s.report.txt", filePrefix)); os.IsNotExist(err) {
 		t.Fatalf("report file not generated during schema-only test")
 	}
-	if _, err := os.Stat(fmt.Sprintf("%sschema.ddl.txt", filePrefix)); os.IsNotExist(err) {
+	if _, err := os.Stat(fmt.Sprintf("%s.schema.ddl.txt", filePrefix)); os.IsNotExist(err) {
 		t.Fatalf("legal ddl file not generated during schema-only test")
 	}
-	if _, err := os.Stat(fmt.Sprintf("%sschema.txt", filePrefix)); os.IsNotExist(err) {
+	if _, err := os.Stat(fmt.Sprintf("%s.schema.txt", filePrefix)); os.IsNotExist(err) {
 		t.Fatalf("readable schema file not generated during schema-only test")
 	}
 	if _, err := os.Stat(sessionFile); os.IsNotExist(err) {
@@ -194,8 +194,8 @@ func TestIntegration_MySQLDUMP_DataSubcommand(t *testing.T) {
 
 	dbName := "test-data-subcommand"
 	dumpFilePath := "../../test_data/mysqldump.test.out"
-	filePrefix := filepath.Join(tmpdir, dbName+".")
-	sessionFile := fmt.Sprintf("%ssession.json", filePrefix)
+	filePrefix := filepath.Join(tmpdir, dbName)
+	sessionFile := fmt.Sprintf("%s.session.json", filePrefix)
 	runSchemaSubcommand(t, dbName, filePrefix, sessionFile, dumpFilePath)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	defer dropDatabase(t, dbURI)
@@ -210,7 +210,7 @@ func TestIntegration_MySQLDUMP_SchemaAndDataSubcommand(t *testing.T) {
 
 	dbName := "test-schema-and-data"
 	dumpFilePath := "../../test_data/mysqldump.test.out"
-	filePrefix := filepath.Join(tmpdir, dbName+".")
+	filePrefix := filepath.Join(tmpdir, dbName)
 
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	runSchemaAndDataSubcommand(t, dbName, dbURI, filePrefix, dumpFilePath)
diff --git a/testing/postgres/integration_test.go b/testing/postgres/integration_test.go
index 19e2244..d1a820c 100644
--- a/testing/postgres/integration_test.go
+++ b/testing/postgres/integration_test.go
@@ -117,7 +117,7 @@ func TestIntegration_PGDUMP_SchemaAndDataSubcommand(t *testing.T) {
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 
 	dataFilepath := "../../test_data/pg_dump.test.out"
-	filePrefix := filepath.Join(tmpdir, dbName+".")
+	filePrefix := filepath.Join(tmpdir, dbName)
 
 	args := fmt.Sprintf("schema-and-data -prefix %s -source=postgres -target-profile='instance=%s,dbName=%s' < %s", filePrefix, instanceID, dbName, dataFilepath)
 	err := common.RunCommand(args, projectID)
@@ -161,7 +161,7 @@ func TestIntegration_POSTGRES_SchemaAndDataSubcommand(t *testing.T) {
 	now := time.Now()
 	dbName, _ := utils.GetDatabaseName(constants.POSTGRES, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	filePrefix := filepath.Join(tmpdir, dbName+".")
+	filePrefix := filepath.Join(tmpdir, dbName)
 
 	args := fmt.Sprintf("schema-and-data -prefix %s -source=postgres -target-profile='instance=%s,dbName=%s'", filePrefix, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
@@ -184,7 +184,7 @@ func TestIntegration_POSTGRES_SchemaSubcommand(t *testing.T) {
 	now := time.Now()
 	dbName, _ := utils.GetDatabaseName(constants.POSTGRES, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	filePrefix := filepath.Join(tmpdir, dbName+".")
+	filePrefix := filepath.Join(tmpdir, dbName)
 
 	args := fmt.Sprintf("schema -prefix %s -source=postgres -target-profile='instance=%s,dbName=%s'", filePrefix, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
diff --git a/webv2/web_test.go b/webv2/web_test.go
index d640cd2..e51d233 100644
--- a/webv2/web_test.go
+++ b/webv2/web_test.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/internal/reports"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
@@ -71,11 +72,11 @@ func TestGetTypeMapPostgres(t *testing.T) {
 	expectedTypemap := map[string][]typeIssue{
 		"bool": {
 			{T: ddl.Bool, DisplayT: ddl.Bool},
-			{T: ddl.Int64, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.Int64, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"bigserial": {
-			{T: ddl.Int64, Brief: internal.IssueDB[internal.Serial].Brief, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief + ", " + internal.IssueDB[internal.Serial].Brief, DisplayT: ddl.String}},
+			{T: ddl.Int64, Brief: reports.IssueDB[internal.Serial].Brief, DisplayT: ddl.Int64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief + ", " + reports.IssueDB[internal.Serial].Brief, DisplayT: ddl.String}},
 		"bpchar": {
 			{T: ddl.Bytes, DisplayT: ddl.Bytes},
 			{T: ddl.String, DisplayT: ddl.String}},
@@ -84,34 +85,34 @@ func TestGetTypeMapPostgres(t *testing.T) {
 			{T: ddl.String, DisplayT: ddl.String}},
 		"date": {
 			{T: ddl.Date, DisplayT: ddl.Date},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"float8": {
 			{T: ddl.Float64, DisplayT: ddl.Float64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"float4": {
-			{T: ddl.Float64, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.Float64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.Float64, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.Float64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"int8": {
 			{T: ddl.Int64, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"int4": {
-			{T: ddl.Int64, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.Int64, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"numeric": {
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
 			{T: ddl.Numeric, DisplayT: ddl.Numeric}},
 		"serial": {
-			{T: ddl.Int64, Brief: internal.IssueDB[internal.Serial].Brief, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief + ", " + internal.IssueDB[internal.Serial].Brief, DisplayT: ddl.String}},
+			{T: ddl.Int64, Brief: reports.IssueDB[internal.Serial].Brief, DisplayT: ddl.Int64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief + ", " + reports.IssueDB[internal.Serial].Brief, DisplayT: ddl.String}},
 		"text": {
 			{T: ddl.Bytes, DisplayT: ddl.Bytes},
 			{T: ddl.String, DisplayT: ddl.String}},
 		"timestamptz": {
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
 			{T: ddl.Timestamp, DisplayT: ddl.Timestamp}},
 		"timestamp": {
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
-			{T: ddl.Timestamp, Brief: internal.IssueDB[internal.Timestamp].Brief, DisplayT: ddl.Timestamp}},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
+			{T: ddl.Timestamp, Brief: reports.IssueDB[internal.Timestamp].Brief, DisplayT: ddl.Timestamp}},
 		"varchar": {
 			{T: ddl.Bytes, DisplayT: ddl.Bytes},
 			{T: ddl.String, DisplayT: ddl.String}},
@@ -166,8 +167,8 @@ func TestGetTypeMapMySQL(t *testing.T) {
 	expectedTypemap := map[string][]typeIssue{
 		"bool": {
 			{T: ddl.Bool, DisplayT: ddl.Bool},
-			{T: ddl.Int64, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.Int64, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"varchar": {
 			{T: ddl.Bytes, DisplayT: ddl.Bytes},
 			{T: ddl.String, DisplayT: ddl.String}},
@@ -186,31 +187,31 @@ func TestGetTypeMapMySQL(t *testing.T) {
 			{T: ddl.Bytes, DisplayT: ddl.Bytes},
 			{T: ddl.String, DisplayT: ddl.String}},
 		"integer": {
-			{T: ddl.Int64, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.Int64, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"smallint": {
-			{T: ddl.Int64, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.Int64, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.Int64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"double": {
 			{T: ddl.Float64, DisplayT: ddl.Float64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"float": {
-			{T: ddl.Float64, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.Float64},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.Float64, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.Float64},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"numeric": {
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
 			{T: ddl.Numeric, DisplayT: ddl.Numeric}},
 		"decimal": {
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
 			{T: ddl.Numeric, DisplayT: ddl.Numeric}},
 		"date": {
 			{T: ddl.Date, DisplayT: ddl.Date},
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String}},
 		"timestamp": {
-			{T: ddl.String, Brief: internal.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Widened].Brief, DisplayT: ddl.String},
 			{T: ddl.Timestamp, DisplayT: ddl.Timestamp}},
 		"time": {
-			{T: ddl.String, Brief: internal.IssueDB[internal.Time].Brief, DisplayT: ddl.String}},
+			{T: ddl.String, Brief: reports.IssueDB[internal.Time].Brief, DisplayT: ddl.String}},
 	}
 	assert.Equal(t, expectedTypemap, typemap)