Repository: cloudspannerecosystem/harbourbridge
Pull request #531 (2023-05-12 07:14:55)

Fix case sensitivity; reserved keyword handling and general stability improvements
This PR fixes a bunch of bugs in Harbourbridge:

- Fix handling of case sensitive source databases when migrating to PGSQL. Now, quoting for PGSQL is source aware and only quotes when Postgres is the source. For non-case sensitive sources like MySQL, no quoting is done.
- Handling of reserved keywords: Irrespective of source database, reserved keywords in PGSQL need to be quoted when the DDL is being generated. For example, if a MySQL database has a table with column name `user`, it needs to be quoted.
- Fix handling of migration of non-public Postgres schemas. This is done by stripping the `tableName` with `schema` at the time of querying.
- QoL improvement - If the postgres database contains only one schema, its name is stripped. If it contains multiple schemas, `public` schema name is stripped.
- Switch default log level to `DEBUG` from `INFO`. This is useful for customers since most errors encountered by the customers can be detected via DEBUG logs.
- Fix log handling for web command. If the web server does not get up, it now logs the error correctly on the UI.
- Add a port flag for the `web` command. Now the user can override the default port from `8080` to a port of their choice.

Note: `InfoSchemaImpl` is passed by value as a receiver in all the method calls inside postgres `infoschema.go`, because of which a bool pointer had to be used to support detection of unique schemas. We would need to refactor the interface + implementation to move all the methods to use pass by reference.  

manitgupta: Addresses issues highlighted in #532, also adds some more stability improvements.

Patch:
diff --git a/.vscode/launch.json b/.vscode/launch.json
index 9b00acc..c000f69 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -33,7 +33,7 @@
             "request": "launch",
             "mode": "debug",
             "program": "${workspaceRoot}/main.go",
-            "args": ["schema-and-data", "-source=XXX", "--target-profile='instance=XXX'", "--source-profile='host=XXX,port=XXX,user=XXX,password=XXX,dbName=XXX'","--log-level=DEBUG"
+            "args": ["schema-and-data", "-source=XXX", "--target-profile=instance=XXX", "--source-profile=host=XXX,port=XXX,user=XXX,password=XXX,dbName=XXX","--log-level=DEBUG"
             ]
         }
     ]
diff --git a/cmd/data.go b/cmd/data.go
index e5d9cb7..55b1e21 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -82,7 +82,7 @@ func (cmd *DataCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
 	f.Int64Var(&cmd.WriteLimit, "write-limit", DefaultWritersLimit, "Write limit for writes to spanner")
 	f.BoolVar(&cmd.dryRun, "dry-run", false, "Flag for generating DDL and schema conversion report without creating a spanner database")
-	f.StringVar(&cmd.logLevel, "log-level", "INFO", "Configure the logging level for the command (INFO, DEBUG), defaults to INFO")
+	f.StringVar(&cmd.logLevel, "log-level", "DEBUG", "Configure the logging level for the command (INFO, DEBUG), defaults to DEBUG")
 	f.BoolVar(&cmd.SkipForeignKeys, "skip-foreign-keys", false, "Skip creating foreign keys after data migration is complete (ddl statements for foreign keys can still be found in the downloaded schema.ddl.txt file and the same can be applied separately)")
 }
 
diff --git a/cmd/schema.go b/cmd/schema.go
index feedd40..e723da7 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -73,7 +73,7 @@ func (cmd *SchemaCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.target, "target", "Spanner", "Specifies the target DB, defaults to Spanner (accepted values: `Spanner`)")
 	f.StringVar(&cmd.targetProfile, "target-profile", "", "Flag for specifying connection profile for target database e.g., \"dialect=postgresql\"")
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
-	f.StringVar(&cmd.logLevel, "log-level", "INFO", "Configure the logging level for the command (INFO, DEBUG), defaults to INFO")
+	f.StringVar(&cmd.logLevel, "log-level", "DEBUG", "Configure the logging level for the command (INFO, DEBUG), defaults to DEBUG")
 	f.BoolVar(&cmd.dryRun, "dry-run", false, "Flag for generating DDL and schema conversion report without creating a spanner database")
 }
 
@@ -111,7 +111,7 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 		return subcommands.ExitFailure
 	}
 
-	conversion.WriteSchemaFile(conv, schemaConversionStartTime, cmd.filePrefix+schemaFile, ioHelper.Out)
+	conversion.WriteSchemaFile(conv, schemaConversionStartTime, cmd.filePrefix+schemaFile, ioHelper.Out, sourceProfile.Driver)
 	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
 
 	// Populate migration request id and migration type in conv object.
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index fda806f..b9a0fc4 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -79,7 +79,7 @@ func (cmd *SchemaAndDataCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.filePrefix, "prefix", "", "File prefix for generated files")
 	f.Int64Var(&cmd.WriteLimit, "write-limit", DefaultWritersLimit, "Write limit for writes to spanner")
 	f.BoolVar(&cmd.dryRun, "dry-run", false, "Flag for generating DDL and schema conversion report without creating a spanner database")
-	f.StringVar(&cmd.logLevel, "log-level", "INFO", "Configure the logging level for the command (INFO, DEBUG), defaults to INFO")
+	f.StringVar(&cmd.logLevel, "log-level", "DEBUG", "Configure the logging level for the command (INFO, DEBUG), defaults to DEBUG")
 }
 
 func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
@@ -127,7 +127,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
 	conv.Audit.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
 
-	conversion.WriteSchemaFile(conv, schemaConversionStartTime, cmd.filePrefix+schemaFile, ioHelper.Out)
+	conversion.WriteSchemaFile(conv, schemaConversionStartTime, cmd.filePrefix+schemaFile, ioHelper.Out, sourceProfile.Driver)
 	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
 	conv.Audit.SkipMetricsPopulation = os.Getenv("SKIP_METRICS_POPULATION") == "true"
 
diff --git a/cmd/utils.go b/cmd/utils.go
index 44aa820..7c22066 100644
--- a/cmd/utils.go
+++ b/cmd/utils.go
@@ -163,7 +163,7 @@ func migrateData(ctx context.Context, targetProfile profiles.TargetProfile, sour
 	}
 	conv.Audit.Progress.UpdateProgress("Data migration complete.", completionPercentage, internal.DataMigrationComplete)
 	if !cmd.SkipForeignKeys {
-		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
+		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out, sourceProfile.Driver); err != nil {
 			err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
 			return bw, err
 		}
@@ -187,7 +187,7 @@ func migrateSchemaAndData(ctx context.Context, targetProfile profiles.TargetProf
 
 	conv.Audit.Progress.UpdateProgress("Data migration complete.", completionPercentage, internal.DataMigrationComplete)
 	if !cmd.SkipForeignKeys {
-		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
+		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out, sourceProfile.Driver); err != nil {
 			err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
 			return bw, err
 		}
diff --git a/conversion/conversion.go b/conversion/conversion.go
index f75502b..f5cff41 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -539,12 +539,12 @@ func CreateOrUpdateDatabase(ctx context.Context, adminClient *database.DatabaseA
 		ctx = metadata.AppendToOutgoingContext(ctx, constants.MigrationMetadataKey, migrationMetadataValue)
 	}
 	if dbExists {
-		err := UpdateDatabase(ctx, adminClient, dbURI, conv, out)
+		err := UpdateDatabase(ctx, adminClient, dbURI, conv, out, driver)
 		if err != nil {
 			return fmt.Errorf("can't update database schema: %v", err)
 		}
 	} else {
-		err := CreateDatabase(ctx, adminClient, dbURI, conv, out)
+		err := CreateDatabase(ctx, adminClient, dbURI, conv, out, driver)
 		if err != nil {
 			return fmt.Errorf("can't create database: %v", err)
 		}
@@ -556,7 +556,7 @@ func CreateOrUpdateDatabase(ctx context.Context, adminClient *database.DatabaseA
 // It automatically determines an appropriate project, selects a
 // Spanner instance to use, generates a new Spanner DB name,
 // and call into the Spanner admin interface to create the new DB.
-func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
+func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File, driver string) error {
 	project, instance, dbName := utils.ParseDbURI(dbURI)
 	fmt.Fprintf(out, "Creating new database %s in instance %s with default permissions ... \n", dbName, instance)
 	// The schema we send to Spanner excludes comments (since Cloud
@@ -575,7 +575,7 @@ func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 		req.DatabaseDialect = adminpb.DatabaseDialect_POSTGRESQL
 	} else {
 		req.CreateStatement = "CREATE DATABASE `" + dbName + "`"
-		req.ExtraStatements = conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: true, ForeignKeys: false, SpDialect: conv.SpDialect})
+		req.ExtraStatements = conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: true, ForeignKeys: false, SpDialect: conv.SpDialect, Source: driver})
 	}
 
 	op, err := adminClient.CreateDatabase(ctx, req)
@@ -589,19 +589,19 @@ func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 
 	if conv.SpDialect == constants.DIALECT_POSTGRESQL {
 		// Update schema separately for PG databases.
-		return UpdateDatabase(ctx, adminClient, dbURI, conv, out)
+		return UpdateDatabase(ctx, adminClient, dbURI, conv, out, driver)
 	}
 	return nil
 }
 
 // UpdateDatabase updates an existing spanner database.
-func UpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
+func UpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File, driver string) error {
 	fmt.Fprintf(out, "Updating schema for %s with default permissions ... \n", dbURI)
 	// The schema we send to Spanner excludes comments (since Cloud
 	// Spanner DDL doesn't accept them), and protects table and col names
 	// using backticks (to avoid any issues with Spanner reserved words).
 	// Foreign Keys are set to false since we create them post data migration.
-	schema := conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: true, ForeignKeys: false, SpDialect: conv.SpDialect})
+	schema := conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: true, ForeignKeys: false, SpDialect: conv.SpDialect, Source: driver})
 	req := &adminpb.UpdateDatabaseDdlRequest{
 		Database:   dbURI,
 		Statements: schema,
@@ -623,11 +623,11 @@ func UpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 
 // UpdateDDLForeignKeys updates the Spanner database with foreign key
 // constraints using ALTER TABLE statements.
-func UpdateDDLForeignKeys(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
+func UpdateDDLForeignKeys(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File, driver string) error {
 	// The schema we send to Spanner excludes comments (since Cloud
 	// Spanner DDL doesn't accept them), and protects table and col names
 	// using backticks (to avoid any issues with Spanner reserved words).
-	fkStmts := conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: false, ForeignKeys: true, SpDialect: conv.SpDialect})
+	fkStmts := conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: false, ForeignKeys: true, SpDialect: conv.SpDialect, Source: driver})
 	if len(fkStmts) == 0 {
 		return nil
 	}
@@ -699,7 +699,7 @@ Recommended value is between 20-30.`)
 // WriteSchemaFile writes DDL statements in a file. It includes CREATE TABLE
 // statements and ALTER TABLE statements to add foreign keys.
 // The parameter name should end with a .txt.
-func WriteSchemaFile(conv *internal.Conv, now time.Time, name string, out *os.File) {
+func WriteSchemaFile(conv *internal.Conv, now time.Time, name string, out *os.File, driver string) {
 	f, err := os.Create(name)
 	if err != nil {
 		fmt.Fprintf(out, "Can't create schema file %s: %v\n", name, err)
@@ -710,7 +710,7 @@ func WriteSchemaFile(conv *internal.Conv, now time.Time, name string, out *os.Fi
 	// and doesn't add backticks around table and column names. This file is
 	// intended for explanatory and documentation purposes, and is not strictly
 	// legal Cloud Spanner DDL (Cloud Spanner doesn't currently support comments).
-	spDDL := conv.SpSchema.GetDDL(ddl.Config{Comments: true, ProtectIds: false, Tables: true, ForeignKeys: true, SpDialect: conv.SpDialect})
+	spDDL := conv.SpSchema.GetDDL(ddl.Config{Comments: true, ProtectIds: false, Tables: true, ForeignKeys: true, SpDialect: conv.SpDialect, Source: driver})
 	if len(spDDL) == 0 {
 		spDDL = []string{"\n-- Schema is empty -- no tables found\n"}
 	}
@@ -737,7 +737,7 @@ func WriteSchemaFile(conv *internal.Conv, now time.Time, name string, out *os.Fi
 
 	// We change 'Comments' to false and 'ProtectIds' to true below to write out a
 	// schema file that is a legal Cloud Spanner DDL.
-	spDDL = conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: true, ForeignKeys: true, SpDialect: conv.SpDialect})
+	spDDL = conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: true, ForeignKeys: true, SpDialect: conv.SpDialect, Source: driver})
 	if len(spDDL) == 0 {
 		spDDL = []string{"\n-- Schema is empty -- no tables found\n"}
 	}
@@ -784,7 +784,7 @@ func WriteConvGeneratedFiles(conv *internal.Conv, dbName string, driver string,
 		return "", err
 	}
 	schemaFileName := dirPath + dbName + "_schema.txt"
-	WriteSchemaFile(conv, now, schemaFileName, out)
+	WriteSchemaFile(conv, now, schemaFileName, out, driver)
 	reportFileName := dirPath + dbName
 	Report(driver, nil, BytesRead, "", conv, reportFileName, dbName, out)
 	sessionFileName := dirPath + dbName + ".session.json"
@@ -967,7 +967,13 @@ func GetInfoSchema(sourceProfile profiles.SourceProfile, targetProfile profiles.
 		if err != nil {
 			return nil, err
 		}
-		return postgres.InfoSchemaImpl{Db: db, SourceProfile: sourceProfile, TargetProfile: targetProfile}, nil
+		temp := false
+		return postgres.InfoSchemaImpl{
+			Db:            db,
+			SourceProfile: sourceProfile,
+			TargetProfile: targetProfile,
+			IsSchemaUnique: &temp, //this is a workaround to set a bool pointer
+		}, nil
 	case constants.DYNAMODB:
 		mySession := session.Must(session.NewSession())
 		dydbClient := dydb.New(mySession, connectionConfig.(*aws.Config))
diff --git a/sources/postgres/infoschema.go b/sources/postgres/infoschema.go
index cf1f6b7..36fc3bf 100644
--- a/sources/postgres/infoschema.go
+++ b/sources/postgres/infoschema.go
@@ -22,6 +22,7 @@ import (
 	"reflect"
 	"sort"
 	"strconv"
+	"strings"
 	"time"
 
 	"cloud.google.com/go/civil"
@@ -38,9 +39,22 @@ import (
 
 // InfoSchemaImpl postgres specific implementation for InfoSchema.
 type InfoSchemaImpl struct {
-	Db            *sql.DB
-	SourceProfile profiles.SourceProfile
-	TargetProfile profiles.TargetProfile
+	Db             *sql.DB
+	SourceProfile  profiles.SourceProfile
+	TargetProfile  profiles.TargetProfile
+	IsSchemaUnique *bool
+}
+
+func (isi InfoSchemaImpl) populateSchemaIsUnique(schemaAndNames []common.SchemaAndName) {
+	schemaSet := make(map[string]struct{})
+	for _, table := range schemaAndNames {
+		schemaSet[table.Schema] = struct{}{}
+	}
+	if len(schemaSet) == 1 {
+		*isi.IsSchemaUnique = true
+	} else {
+		*isi.IsSchemaUnique = false
+	}
 }
 
 // StartChangeDataCapture is used for automatic triggering of Datastream job when
@@ -76,7 +90,9 @@ func (isi InfoSchemaImpl) GetToDdl() common.ToDdl {
 
 // GetTableName returns table name.
 func (isi InfoSchemaImpl) GetTableName(schema string, tableName string) string {
-	if schema == "public" { // Drop 'public' prefix.
+	if *isi.IsSchemaUnique { // Drop schema name as prefix if only one schema is detected.
+		return tableName
+	} else if schema == "public" {
 		return tableName
 	}
 	return fmt.Sprintf("%s.%s", schema, tableName)
@@ -87,7 +103,14 @@ func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, tableId string)
 	// PostgreSQL schema and name can be arbitrary strings.
 	// Ideally we would pass schema/name as a query parameter,
 	// but PostgreSQL doesn't support this. So we quote it instead.
-	q := fmt.Sprintf(`SELECT * FROM "%s"."%s";`, conv.SrcSchema[tableId].Schema, conv.SrcSchema[tableId].Name)
+	isSchemaNamePrefixed := strings.HasPrefix(conv.SrcSchema[tableId].Name, conv.SrcSchema[tableId].Schema+".")
+	var tableName string
+	if isSchemaNamePrefixed {
+		tableName = strings.TrimPrefix(conv.SrcSchema[tableId].Name, conv.SrcSchema[tableId].Schema+".")
+	} else {
+		tableName = conv.SrcSchema[tableId].Name
+	}
+	q := fmt.Sprintf(`SELECT * FROM "%s"."%s";`, conv.SrcSchema[tableId].Schema, tableName)
 	rows, err := isi.Db.Query(q)
 	if err != nil {
 		return nil, err
@@ -231,6 +254,7 @@ func (isi InfoSchemaImpl) GetTables() ([]common.SchemaAndName, error) {
 			tables = append(tables, common.SchemaAndName{Schema: tableSchema, Name: tableName})
 		}
 	}
+	isi.populateSchemaIsUnique(tables)
 	return tables, nil
 }
 
diff --git a/spanner/ddl/ast.go b/spanner/ddl/ast.go
index ca21ef8..3d30fe7 100644
--- a/spanner/ddl/ast.go
+++ b/spanner/ddl/ast.go
@@ -91,6 +91,16 @@ var PGSQL_TO_STANDARD_TYPE_TYPEMAP = map[string]string{
 	PGJSONB:       JSON,
 }
 
+//List taken from - https://www.postgresql.org/docs/current/sql-keywords-appendix.html
+//Assumption is that this list PGSQL dialect uses the same keywords
+var PGSQL_RESERVED_KEYWORD_LIST = []string{"ALL", "ANALYSE", "ANALYZE", "AND", "ANY", "ASC",
+	"ASYMMETRIC", "BOTH", "CASE", "CAST", "CHECK", "COLLATE", "COLUMN", "CONSTRAINT", "CURRENT_CATALOG",
+	"CURRENT_DATE", "CURRENT_ROLE", "CURRENT_TIME", "CURRENT_TIMESTAMP", "CURRENT_USER", "DEFAULT",
+	"DEFERRABLE", "DESC", "DISTINCT", "DO", "ELSE", "END", "FALSE", "FOREIGN", "IN", "INITIALLY",
+	"LATERAL", "LEADING", "LOCALTIME", "LOCALTIMESTAMP", "NOT", "NULL", "ONLY", "OR", "PLACING",
+	"PRIMARY", "REFERENCES", "SELECT", "SESSION_USER", "SOME", "SYMMETRIC", "TABLE", "THEN",
+	"TRAILING", "TRUE", "UNIQUE", "USER", "USING", "VARIADIC", "WHEN"}
+
 // Type represents the type of a column.
 //
 //	type:
@@ -173,12 +183,35 @@ type Config struct {
 	Tables      bool // If true, print tables
 	ForeignKeys bool // If true, print foreign key constraints.
 	SpDialect   string
+	Source		string //SourceDB information for determining case-sensitivity handling for PGSQL
+}
+
+func isIdentifierReservedInPG(identifier string) bool {
+	for _, KEYWORD := range PGSQL_RESERVED_KEYWORD_LIST {
+		if strings.EqualFold(KEYWORD, identifier) {
+			return true
+		}
+	}
+	return false
+}
+
+func isSourceCaseSensitive(source string) bool {
+	switch source {
+	case constants.POSTGRES, constants.PGDUMP:
+		return true
+	default:
+		return false
+	}
 }
 
 func (c Config) quote(s string) string {
 	if c.ProtectIds {
 		if c.SpDialect == constants.DIALECT_POSTGRESQL {
-			return s
+			if isIdentifierReservedInPG(s) || isSourceCaseSensitive(c.Source) {
+				return "\"" + s + "\""
+			} else {
+				return s
+			}
 		} else {
 			return "`" + s + "`"
 		}
diff --git a/webv2/table/review_table_schema.go b/webv2/table/review_table_schema.go
index 8022352..ac84c8b 100644
--- a/webv2/table/review_table_schema.go
+++ b/webv2/table/review_table_schema.go
@@ -129,7 +129,7 @@ func ReviewTableSchema(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	ddl := GetSpannerTableDDL(conv.SpSchema[tableId], conv.SpDialect)
+	ddl := GetSpannerTableDDL(conv.SpSchema[tableId], conv.SpDialect, sessionState.Driver)
 
 	interleaveTableSchema = trimRedundantInterleaveTableSchema(interleaveTableSchema)
 	// update interleaveTableSchema by filling the missing fields.
diff --git a/webv2/table/utilities.go b/webv2/table/utilities.go
index 50af0cd..c6fccec 100644
--- a/webv2/table/utilities.go
+++ b/webv2/table/utilities.go
@@ -40,9 +40,9 @@ func IsColumnPresentInColNames(colIds []string, colId string) bool {
 }
 
 // GetSpannerTableDDL return Spanner Table DDL as string.
-func GetSpannerTableDDL(spannerTable ddl.CreateTable, spDialect string) string {
+func GetSpannerTableDDL(spannerTable ddl.CreateTable, spDialect string, driver string) string {
 	sessionState := session.GetSessionState()
-	c := ddl.Config{Comments: true, ProtectIds: false, SpDialect: spDialect}
+	c := ddl.Config{Comments: true, ProtectIds: false, SpDialect: spDialect, Source: driver}
 
 	ddl := spannerTable.PrintCreateTable(sessionState.Conv.SpSchema, c)
 
diff --git a/webv2/web.go b/webv2/web.go
index a64f9fc..cfdb390 100644
--- a/webv2/web.go
+++ b/webv2/web.go
@@ -242,7 +242,8 @@ func convertSchemaSQL(w http.ResponseWriter, r *http.Request) {
 	case constants.MYSQL:
 		err = common.ProcessSchema(conv, mysql.InfoSchemaImpl{DbName: sessionState.DbName, Db: sessionState.SourceDB}, common.DefaultWorkers)
 	case constants.POSTGRES:
-		err = common.ProcessSchema(conv, postgres.InfoSchemaImpl{Db: sessionState.SourceDB}, common.DefaultWorkers)
+		temp := false
+		err = common.ProcessSchema(conv, postgres.InfoSchemaImpl{Db: sessionState.SourceDB, IsSchemaUnique: &temp}, common.DefaultWorkers)
 	case constants.SQLSERVER:
 		err = common.ProcessSchema(conv, sqlserver.InfoSchemaImpl{DbName: sessionState.DbName, Db: sessionState.SourceDB}, common.DefaultWorkers)
 	case constants.ORACLE:
@@ -544,7 +545,7 @@ func fetchLastLoadedSessionDetails(w http.ResponseWriter, r *http.Request) {
 // build DDL to send to Spanner.
 func getDDL(w http.ResponseWriter, r *http.Request) {
 	sessionState := session.GetSessionState()
-	c := ddl.Config{Comments: true, ProtectIds: false, SpDialect: sessionState.Conv.SpDialect}
+	c := ddl.Config{Comments: true, ProtectIds: false, SpDialect: sessionState.Conv.SpDialect, Source: sessionState.Driver}
 	var tables []string
 	for t := range sessionState.Conv.SpSchema {
 		tables = append(tables, t)
@@ -893,7 +894,7 @@ func getSchemaFile(w http.ResponseWriter, r *http.Request) {
 	schemaFileName := "frontend/" + filePrefix + "schema.txt"
 
 	sessionState := session.GetSessionState()
-	conversion.WriteSchemaFile(sessionState.Conv, now, schemaFileName, ioHelper.Out)
+	conversion.WriteSchemaFile(sessionState.Conv, now, schemaFileName, ioHelper.Out, sessionState.Driver)
 	schemaAbsPath, err := filepath.Abs(schemaFileName)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Can not create absolute path : %v", err), http.StatusInternalServerError)
@@ -2272,16 +2273,16 @@ func init() {
 }
 
 // App connects to the web app v2.
-func App(logLevel string, open bool) {
+func App(logLevel string, open bool, port int) error {
 	err := logger.InitializeLogger(logLevel)
 	if err != nil {
-		log.Fatal("Error initialising webapp, did you specify a valid log-level? [DEBUG, INFO, WARN, ERROR, FATAL]")
+		return fmt.Errorf("error initialising webapp, did you specify a valid log-level? [DEBUG, INFO]")
 	}
-	addr := ":8080"
+	addr := fmt.Sprintf(":%s",strconv.Itoa(port))
 	router := getRoutes()
-	fmt.Println("Harbourbridge UI started at:", fmt.Sprintf("http://localhost%s", addr))
+	fmt.Println("Starting Harbourbridge UI at:", fmt.Sprintf("http://localhost%s", addr))
 	if open {
 		browser.OpenURL(fmt.Sprintf("http://localhost%s", addr))
 	}
-	log.Fatal(http.ListenAndServe(addr, handlers.CORS(handlers.AllowedHeaders([]string{"X-Requested-With", "Content-Type", "Authorization"}), handlers.AllowedMethods([]string{"GET", "POST", "PUT", "HEAD", "OPTIONS"}), handlers.AllowedOrigins([]string{"*"}))(router)))
+	return http.ListenAndServe(addr, handlers.CORS(handlers.AllowedHeaders([]string{"X-Requested-With", "Content-Type", "Authorization"}), handlers.AllowedMethods([]string{"GET", "POST", "PUT", "HEAD", "OPTIONS"}), handlers.AllowedOrigins([]string{"*"}))(router))
 }
diff --git a/webv2/webCmd.go b/webv2/webCmd.go
index 1e6e63c..99995bd 100644
--- a/webv2/webCmd.go
+++ b/webv2/webCmd.go
@@ -24,9 +24,7 @@ import (
 	"path/filepath"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-	"github.com/cloudspannerecosystem/harbourbridge/logger"
 	"github.com/google/subcommands"
-	"go.uber.org/zap"
 )
 
 var FrontendDir embed.FS
@@ -35,6 +33,7 @@ type WebCmd struct {
 	DistDir  embed.FS
 	logLevel string
 	open     bool
+	port	int
 }
 
 // Name returns the name of operation.
@@ -52,8 +51,9 @@ func (cmd *WebCmd) Usage() string {
 }
 
 func (cmd *WebCmd) SetFlags(f *flag.FlagSet) {
-	f.StringVar(&cmd.logLevel, "log-level", "INFO", "Configure the logging level for the command (INFO, DEBUG), defaults to INFO")
+	f.StringVar(&cmd.logLevel, "log-level", "DEBUG", "Configure the logging level for the command (INFO, DEBUG), defaults to DEBUG")
 	f.BoolVar(&cmd.open, "open", false, "Opens the Harbourbridge web interface in the default browser, defaults to false")
+	f.IntVar(&cmd.port, "port", 8080, "The port in which Harbourbridge will run, defaults to 8080")
 }
 
 func (cmd *WebCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
@@ -62,10 +62,9 @@ func (cmd *WebCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{
 	var err error
 	defer func() {
 		if err != nil {
-			logger.Log.Fatal("FATAL error", zap.Error(err))
+			fmt.Printf("FATAL error, unable to start webapp: %s", err)
 		}
 	}()
-	defer logger.Log.Sync()
-	App(cmd.logLevel, cmd.open)
+	err = App(cmd.logLevel, cmd.open, cmd.port)
 	return subcommands.ExitSuccess
 }


Test patch:
diff --git a/sources/postgres/infoschema_test.go b/sources/postgres/infoschema_test.go
index 1ae5540..2ac0010 100644
--- a/sources/postgres/infoschema_test.go
+++ b/sources/postgres/infoschema_test.go
@@ -229,7 +229,7 @@ func TestProcessSchema(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessSchema(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}}, 1)
+	err := common.ProcessSchema(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}, newFalsePtr()}, 1)
 	assert.Nil(t, err)
 	expectedSchema := map[string]ddl.CreateTable{
 		"user": ddl.CreateTable{
@@ -363,7 +363,7 @@ func TestProcessData(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessData(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}})
+	common.ProcessData(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}, newFalsePtr()})
 
 	assert.Equal(t,
 		[]spannerData{
@@ -504,7 +504,7 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 	}
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
-	err := common.ProcessSchema(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}}, 1)
+	err := common.ProcessSchema(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}, newFalsePtr()}, 1)
 	assert.Nil(t, err)
 	conv.SetDataMode()
 	var rows []spannerData
@@ -512,7 +512,7 @@ func TestConvertSqlRow_MultiCol(t *testing.T) {
 		func(table string, cols []string, vals []interface{}) {
 			rows = append(rows, spannerData{table: table, cols: cols, vals: vals})
 		})
-	common.ProcessData(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}})
+	common.ProcessData(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}, newFalsePtr()})
 	assert.Equal(t, []spannerData{
 		{table: "test", cols: []string{"a", "b", "synth_id"}, vals: []interface{}{"cat", float64(42.3), "0"}},
 		{table: "test", cols: []string{"a", "c", "synth_id"}, vals: []interface{}{"dog", int64(22), "-9223372036854775808"}}},
@@ -539,7 +539,7 @@ func TestSetRowStats(t *testing.T) {
 	db := mkMockDB(t, ms)
 	conv := internal.MakeConv()
 	conv.SetDataMode()
-	common.SetRowStats(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}})
+	common.SetRowStats(conv, InfoSchemaImpl{db, profiles.SourceProfile{}, profiles.TargetProfile{}, newFalsePtr()})
 	assert.Equal(t, int64(5), conv.Stats.Rows["test1"])
 	assert.Equal(t, int64(142), conv.Stats.Rows["test2"])
 	assert.Equal(t, int64(0), conv.Unexpecteds())
@@ -562,3 +562,8 @@ func mkMockDB(t *testing.T, ms []mockSpec) *sql.DB {
 	}
 	return db
 }
+
+func newFalsePtr() *bool {
+	temp := false
+	return &temp
+}
diff --git a/testing/conversion/conversion_test.go b/testing/conversion/conversion_test.go
index d68e8d3..de43010 100644
--- a/testing/conversion/conversion_test.go
+++ b/testing/conversion/conversion_test.go
@@ -195,12 +195,12 @@ func TestUpdateDDLForeignKeys(t *testing.T) {
 	for _, tc := range testCases {
 		dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, tc.dbName)
 		conv := BuildConv(t, tc.numCols, tc.numFks, false)
-		err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, conv, os.Stdout)
+		err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, conv, os.Stdout, "")
 		if err != nil {
 			t.Fatal(err)
 		}
 		conversion.MaxWorkers = tc.numWorkers
-		if err = conversion.UpdateDDLForeignKeys(ctx, databaseAdmin, dbURI, conv, os.Stdout); err != nil {
+		if err = conversion.UpdateDDLForeignKeys(ctx, databaseAdmin, dbURI, conv, os.Stdout, ""); err != nil {
 			t.Fatalf("\nCan't perform update operation on db %s with foreign keys: %v\n", tc.dbName, err)
 		}
 
@@ -226,7 +226,7 @@ func TestVerifyDb(t *testing.T) {
 	for _, tc := range testCases {
 		dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, tc.dbName)
 		if tc.dbExists {
-			err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout)
+			err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout, "")
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -250,7 +250,7 @@ func TestVerifyDb(t *testing.T) {
 func TestCheckExistingDb(t *testing.T) {
 	onlyRunForEmulatorTest(t)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, "check-db-exists")
-	err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, internal.MakeConv(), os.Stdout)
+	err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, internal.MakeConv(), os.Stdout, "")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -283,7 +283,7 @@ func TestValidateDDL(t *testing.T) {
 
 	for _, tc := range testCases {
 		dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, tc.dbName)
-		err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout)
+		err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout, "")
 		if err != nil {
 			t.Fatal(err)
 		}
diff --git a/webv2/table/review_table_schema_test.go b/webv2/table/review_table_schema_test.go
index 92acbf6..8a586ed 100644
--- a/webv2/table/review_table_schema_test.go
+++ b/webv2/table/review_table_schema_test.go
@@ -368,7 +368,7 @@ func TestReviewTableSchema(t *testing.T) {
 				status, tc.statusCode)
 		}
 
-		expectedddl := GetSpannerTableDDL(tc.expectedConv.SpSchema[tc.tableId], tc.expectedConv.SpDialect)
+		expectedddl := GetSpannerTableDDL(tc.expectedConv.SpSchema[tc.tableId], tc.expectedConv.SpDialect, sessionState.Driver)
 
 		if tc.statusCode == http.StatusOK {
 			assert.Equal(t, expectedddl, res.DDL)