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)
	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")
		f = out
	} else {
		defer f.Close()
	}
	w := bufio.NewWriter(f)
	w.WriteString(banner)
	summary := internal.GenerateReport(driver, conv, w, badWrites, true, true)
	w.Flush()
	var isDump bool
	if strings.Contains(driver, "dump") {
		isDump = true
	}
	if isDump {
		fmt.Fprintf(out, "Processed %d bytes of %s data (%d statements, %d rows of data, %d errors, %d unexpected conditions).\n",
			BytesRead, driver, conv.Statements(), conv.Rows(), conv.StatementErrors(), conv.Unexpecteds())
	} else {
		fmt.Fprintf(out, "Processed source database via %s driver (%d rows of data, %d unexpected conditions).\n",
			driver, conv.Rows(), conv.Unexpecteds())
	}
	if f != out {
		fmt.Fprint(out, summary)
		fmt.Fprintf(out, "See file '%s' for details of the schema and data conversions.\n", reportFileName)
	}
}
Report generates a report of schema and data conversion.
package conversion
import (
	"bufio"
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"os/exec"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
	sp "cloud.google.com/go/spanner"
	database "cloud.google.com/go/spanner/admin/database/apiv1"
	instance "cloud.google.com/go/spanner/admin/instance/apiv1"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	dydb "github.com/aws/aws-sdk-go/service/dynamodb"
	_ "github.com/go-sql-driver/mysql"
	_ "github.com/lib/pq"
	"golang.org/x/crypto/ssh/terminal"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"
	adminpb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
	instancepb "google.golang.org/genproto/googleapis/spanner/admin/instance/v1"
	"github.com/cloudspannerecosystem/harbourbridge/internal"
	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
	"github.com/cloudspannerecosystem/harbourbridge/sources/dynamodb"
	"github.com/cloudspannerecosystem/harbourbridge/sources/mysql"
	"github.com/cloudspannerecosystem/harbourbridge/sources/postgres"
	"github.com/cloudspannerecosystem/harbourbridge/spanner"
	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
)
const (
	PGDUMP string = "pg_dump"
	POSTGRES string = "postgres"
	MYSQLDUMP string = "mysqldump"
	MYSQL string = "mysql"
	DYNAMODB string = "dynamodb"
	TARGET_SPANNER               string = "spanner"
	TARGET_EXPERIMENTAL_POSTGRES string = "experimental_postgres"
)
var (
	MaxWorkers = 20
)
func SchemaConv(driver string, targetDb string, ioHelper *IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
	switch driver {
	case POSTGRES, MYSQL:
		return schemaFromSQL(driver, targetDb)
	case PGDUMP, MYSQLDUMP:
		return schemaFromDump(driver, targetDb, ioHelper)
	case DYNAMODB:
		return schemaFromDynamoDB(schemaSampleSize)
	default:
		return nil, fmt.Errorf("schema conversion for driver %s not supported", driver)
	}
}
func DataConv(driver string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
	config := spanner.BatchWriterConfig{
		BytesLimit: 100 * 1000 * 1000,
		WriteLimit: 40,
		RetryLimit: 1000,
		Verbose:    internal.Verbose(),
	}
	switch driver {
	case POSTGRES, MYSQL:
		return dataFromSQL(driver, config, client, conv)
	case PGDUMP, MYSQLDUMP:
		if conv.SpSchema.CheckInterleaved() {
			return nil, fmt.Errorf("HarbourBridge does not currently support data conversion from dump files\nif the schema contains interleaved tables. Suggest using direct access to source database\ni.e. using drivers postgres and mysql.")
		}
		return dataFromDump(driver, config, ioHelper, client, conv, dataOnly)
	case DYNAMODB:
		return dataFromDynamoDB(config, client, conv)
	default:
		return nil, fmt.Errorf("data conversion for driver %s not supported", driver)
	}
}
func driverConfig(driver string) (string, error) {
	switch driver {
	case POSTGRES:
		return pgDriverConfig()
	case MYSQL:
		return mysqlDriverConfig()
	default:
		return "", fmt.Errorf("Driver %s not supported", driver)
	}
}
func pgDriverConfig() (string, error) {
	server := os.Getenv("PGHOST")
	port := os.Getenv("PGPORT")
	user := os.Getenv("PGUSER")
	dbname := os.Getenv("PGDATABASE")
	if server == "" || port == "" || user == "" || dbname == "" {
		fmt.Printf("Please specify host, port, user and database using PGHOST, PGPORT, PGUSER and PGDATABASE environment variables\n")
		return "", fmt.Errorf("Could not connect to source database")
	}
	password := os.Getenv("PGPASSWORD")
	if password == "" {
		password = getPassword()
	}
	return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbname), nil
}
func mysqlDriverConfig() (string, error) {
	server := os.Getenv("MYSQLHOST")
	port := os.Getenv("MYSQLPORT")
	user := os.Getenv("MYSQLUSER")
	dbname := os.Getenv("MYSQLDATABASE")
	if server == "" || port == "" || user == "" || dbname == "" {
		fmt.Printf("Please specify host, port, user and database using MYSQLHOST, MYSQLPORT, MYSQLUSER and MYSQLDATABASE environment variables\n")
		return "", fmt.Errorf("Could not connect to source database")
	}
	password := os.Getenv("MYSQLPWD")
	if password == "" {
		password = getPassword()
	}
	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbname), nil
}
func schemaFromSQL(driver string, targetDb string) (*internal.Conv, error) {
	driverConfig, err := driverConfig(driver)
	if err != nil {
		return nil, err
	}
	sourceDB, err := sql.Open(driver, driverConfig)
	if err != nil {
		return nil, err
	}
	conv := internal.MakeConv()
	conv.TargetDb = targetDb
	err = ProcessInfoSchema(driver, conv, sourceDB)
	if err != nil {
		return nil, err
	}
	return conv, nil
}
func dataFromSQL(driver string, config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
	driverConfig, err := driverConfig(driver)
	if err != nil {
		return nil, err
	}
	sourceDB, err := sql.Open(driver, driverConfig)
	if err != nil {
		return nil, err
	}
	err = SetRowStats(driver, conv, sourceDB)
	if err != nil {
		return nil, err
	}
	totalRows := conv.Rows()
	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
	rows := int64(0)
	config.Write = func(m []*sp.Mutation) error {
		_, err := client.Apply(context.Background(), m)
		if err != nil {
			return err
		}
		atomic.AddInt64(&rows, int64(len(m)))
		p.MaybeReport(atomic.LoadInt64(&rows))
		return nil
	}
	writer := spanner.NewBatchWriter(config)
	conv.SetDataMode()
	conv.SetDataSink(
		func(table string, cols []string, vals []interface{}) {
			writer.AddRow(table, cols, vals)
		})
	err = ProcessSQLData(driver, conv, sourceDB)
	if err != nil {
		return nil, err
	}
	writer.Flush()
	return writer, nil
}
func getDynamoDBClientConfig() *aws.Config {
	cfg := aws.Config{}
	endpointOverride := os.Getenv("DYNAMODB_ENDPOINT_OVERRIDE")
	if endpointOverride != "" {
		cfg.Endpoint = aws.String(endpointOverride)
	}
	return &cfg
}
func schemaFromDynamoDB(sampleSize int64) (*internal.Conv, error) {
	conv := internal.MakeConv()
	mySession := session.Must(session.NewSession())
	dydbClient := dydb.New(mySession, getDynamoDBClientConfig())
	err := dynamodb.ProcessSchema(conv, dydbClient, []string{}, sampleSize)
	if err != nil {
		return nil, err
	}
	return conv, nil
}
func dataFromDynamoDB(config spanner.BatchWriterConfig, client *sp.Client, conv *internal.Conv) (*spanner.BatchWriter, error) {
	mySession := session.Must(session.NewSession())
	dydbClient := dydb.New(mySession, getDynamoDBClientConfig())
	dynamodb.SetRowStats(conv, dydbClient)
	totalRows := conv.Rows()
	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
	rows := int64(0)
	config.Write = func(m []*sp.Mutation) error {
		_, err := client.Apply(context.Background(), m)
		if err != nil {
			return err
		}
		atomic.AddInt64(&rows, int64(len(m)))
		p.MaybeReport(atomic.LoadInt64(&rows))
		return nil
	}
	writer := spanner.NewBatchWriter(config)
	conv.SetDataMode()
	conv.SetDataSink(
		func(table string, cols []string, vals []interface{}) {
			writer.AddRow(table, cols, vals)
		})
	err := dynamodb.ProcessData(conv, dydbClient)
	if err != nil {
		return nil, err
	}
	writer.Flush()
	return writer, nil
}
type IOStreams struct {
	In, SeekableIn, Out *os.File
	BytesRead           int64
}
func NewIOStreams(driver string, dumpFile string) IOStreams {
	io := IOStreams{In: os.Stdin, Out: os.Stdout}
	if (driver == PGDUMP || driver == MYSQLDUMP) && dumpFile != "" {
		fmt.Printf("\nLoading dump file from path: %s\n", dumpFile)
		f, err := os.Open(dumpFile)
		if err != nil {
			fmt.Printf("\nError reading dump file: %v err:%v\n", dumpFile, err)
			log.Fatal(err)
		}
		io.In = f
	}
	return io
}
func schemaFromDump(driver string, targetDb string, ioHelper *IOStreams) (*internal.Conv, error) {
	f, n, err := getSeekable(ioHelper.In)
	if err != nil {
		printSeekError(driver, err, ioHelper.Out)
		return nil, fmt.Errorf("can't get seekable input file")
	}
	ioHelper.SeekableIn = f
	ioHelper.BytesRead = n
	conv := internal.MakeConv()
	conv.TargetDb = targetDb
	p := internal.NewProgress(n, "Generating schema", internal.Verbose(), false)
	r := internal.NewReader(bufio.NewReader(f), p)
	conv.SetSchemaMode()  
	conv.SetDataSink(nil)
	err = ProcessDump(driver, conv, r)
	if err != nil {
		fmt.Fprintf(ioHelper.Out, "Failed to parse the data file: %v", err)
		return nil, fmt.Errorf("failed to parse the data file")
	}
	p.Done()
	return conv, nil
}
func dataFromDump(driver string, config spanner.BatchWriterConfig, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
	if !dataOnly {
		_, err := ioHelper.SeekableIn.Seek(0, 0)
		if err != nil {
			fmt.Printf("\nCan't seek to start of file (preparation for second pass): %v\n", err)
			return nil, fmt.Errorf("can't seek to start of file")
		}
	} else {
		f, n, err := getSeekable(ioHelper.In)
		if err != nil {
			printSeekError(driver, err, ioHelper.Out)
			return nil, fmt.Errorf("can't get seekable input file")
		}
		ioHelper.SeekableIn = f
		ioHelper.BytesRead = n
	}
	totalRows := conv.Rows()
	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
	r := internal.NewReader(bufio.NewReader(ioHelper.SeekableIn), nil)
	rows := int64(0)
	config.Write = func(m []*sp.Mutation) error {
		_, err := client.Apply(context.Background(), m)
		if err != nil {
			return err
		}
		atomic.AddInt64(&rows, int64(len(m)))
		p.MaybeReport(atomic.LoadInt64(&rows))
		return nil
	}
	writer := spanner.NewBatchWriter(config)
	conv.SetDataMode()  
	conv.SetDataSink(
		func(table string, cols []string, vals []interface{}) {
			writer.AddRow(table, cols, vals)
		})
	ProcessDump(driver, conv, r)
	writer.Flush()
	p.Done()
	return writer, nil
}