Repository: cloudspannerecosystem/harbourbridge
Pull request #347 (2022-07-19 13:14:27)

Changes to shard processing in DynamoDB Streams
**Issue**: 
- Child shards ended up waiting for parent shards infinitely in case parent shards get deleted within DynamoDB Streams. 

**Things changed**
- Changed the scanning strategy of shards to handle issue.


Patch:
diff --git a/README.md b/README.md
index 19cb6c4..a495aae 100644
--- a/README.md
+++ b/README.md
@@ -352,6 +352,8 @@ conversion state endcoded as JSON.
 
 `-target-profile` Specifies detailed parameters for the target database. See [Target Profile](#target-profile) for details.
 
+`-dry-run` Controls whether we run the migration in dry run mode or not. Using this mode generates schema and report for schema and/or data conversion without any actual creation of tables.
+
 ### Source Profile
 
 HarbourBridge accepts the following params for --source-profile,
diff --git a/cmd/cmd.go b/cmd/cmd.go
index a0c59a9..7896e7b 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -77,6 +77,9 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 		conversion.WriteSchemaFile(conv, now, outputFilePrefix+schemaFile, ioHelper.Out)
 		conversion.WriteSessionFile(conv, outputFilePrefix+sessionFile, ioHelper.Out)
 		if schemaOnly {
+			// Populate migration request id and migration type in conv object.
+			conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
+			conv.Audit.MigrationType = migration.MigrationData_SCHEMA_ONLY.Enum()
 			conversion.Report(driver, nil, ioHelper.BytesRead, "", conv, outputFilePrefix+reportFile, ioHelper.Out)
 			return nil
 		}
@@ -88,7 +91,7 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 		}
 	}
 
-	// Populate migration request id and migration type in conv object
+	// Populate migration request id and migration type in conv object.
 	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
 	if dataOnly {
 		conv.Audit.MigrationType = migration.MigrationData_DATA_ONLY.Enum()
diff --git a/cmd/data.go b/cmd/data.go
index 4f9fbc2..167d28c 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -22,13 +22,15 @@ import (
 	"path"
 	"time"
 
+	sp "cloud.google.com/go/spanner"
+	database "cloud.google.com/go/spanner/admin/database/apiv1"
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/logger"
-	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
 	"github.com/google/subcommands"
 	"github.com/google/uuid"
 	"go.uber.org/zap"
@@ -40,12 +42,12 @@ type DataCmd struct {
 	sourceProfile   string
 	target          string
 	targetProfile   string
-	skipForeignKeys bool
 	sessionJSON     string
 	filePrefix      string // TODO: move filePrefix to global flags
 	writeLimit      int64
 	dryRun          bool
 	logLevel        string
+	skipForeignKeys bool
 }
 
 // Name returns the name of operation.
@@ -76,11 +78,11 @@ func (cmd *DataCmd) SetFlags(f *flag.FlagSet) {
 	f.StringVar(&cmd.sessionJSON, "session", "", "Specifies the file we restore session state from")
 	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.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)")
 	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, "To validate the syntax of the command by running it in an air-gapped manner, such that no network calls are made.")
+	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.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)")
 }
 
 func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
@@ -92,10 +94,6 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 			logger.Log.Fatal("FATAL error", zap.Error(err))
 		}
 	}()
-	if cmd.dryRun {
-		fmt.Print("--dry-run flag is not implemented")
-		return subcommands.ExitFailure
-	}
 	err = logger.InitializeLogger(cmd.logLevel)
 	if err != nil {
 		fmt.Println("Error initialising logger, did you specify a valid log-level? [DEBUG, INFO, WARN, ERROR, FATAL]", err)
@@ -104,54 +102,24 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 	defer logger.Log.Sync()
 
 	conv := internal.MakeConv()
-
-	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
+	sourceProfile, targetProfile, ioHelper, dbName, err := PrepareMigrationPrerequisites(cmd.sourceProfile, cmd.targetProfile, cmd.source)
 	if err != nil {
+		err = fmt.Errorf("error while preparing prerequisites for migration: %v", err)
 		return subcommands.ExitUsageError
 	}
-	sourceProfile.Driver, err = sourceProfile.ToLegacyDriver(cmd.source)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-
-	targetProfile, err := profiles.NewTargetProfile(cmd.targetProfile)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-	targetProfile.TargetDb = targetProfile.ToLegacyTargetDb()
-
-	dumpFilePath := ""
-	if sourceProfile.Ty == profiles.SourceProfileTypeFile && (sourceProfile.File.Format == "" || sourceProfile.File.Format == "dump") {
-		dumpFilePath = sourceProfile.File.Path
-	}
-	ioHelper := utils.NewIOStreams(sourceProfile.Driver, dumpFilePath)
-	if ioHelper.SeekableIn != nil {
-		defer ioHelper.In.Close()
-	}
-
-	now := time.Now()
-	project, instance, dbName, err := targetProfile.GetResourceIds(ctx, now, sourceProfile.Driver, ioHelper.Out)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-	fmt.Println("Using Google Cloud project:", project)
-	fmt.Println("Using Cloud Spanner instance:", instance)
-	utils.PrintPermissionsWarning(sourceProfile.Driver, ioHelper.Out)
-
-	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName)
+	var (
+		bw     *writer.BatchWriter
+		banner string
+	)
+	// Populate migration request id and migration type in conv object.
+	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
+	conv.Audit.MigrationType = migration.MigrationData_DATA_ONLY.Enum()
+	dataCoversionStartTime := time.Now()
 
 	// If filePrefix not explicitly set, use dbName as prefix.
 	if cmd.filePrefix == "" {
 		cmd.filePrefix = dbName + "."
 	}
-
-	client, err := utils.GetClient(ctx, dbURI)
-	if err != nil {
-		err = fmt.Errorf("can't create client for db %s: %v", dbURI, err)
-		return subcommands.ExitFailure
-	}
-	defer client.Close()
-
 	if !sourceProfile.UseTargetSchema() {
 		err = conversion.ReadSessionFile(conv, cmd.sessionJSON)
 		if err != nil {
@@ -163,46 +131,87 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		}
 	}
 
-	adminClient, err := utils.NewDatabaseAdminClient(ctx)
-	if err != nil {
-		err = fmt.Errorf("can't create admin client: %w", utils.AnalyzeError(err, dbURI))
-		return subcommands.ExitFailure
-	}
-	defer adminClient.Close()
-
-	// Populate migration request id and migration type in conv object
-	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
-	conv.Audit.MigrationType = migration.MigrationData_DATA_ONLY.Enum()
-
-	if !sourceProfile.UseTargetSchema() {
-		err = conversion.CreateOrUpdateDatabase(ctx, adminClient, dbURI, sourceProfile.Driver, targetProfile.TargetDb, conv, ioHelper.Out)
+	var (
+		dbURI       string
+		adminClient *database.DatabaseAdminClient
+		client      *sp.Client
+	)
+	if !cmd.dryRun {
+		now := time.Now()
+		adminClient, client, dbURI, err = CreateDatabaseClient(ctx, targetProfile, sourceProfile.Driver, ioHelper)
 		if err != nil {
-			err = fmt.Errorf("can't create/update database: %v", err)
+			err = fmt.Errorf("can't create database client: %v", err)
+			return subcommands.ExitFailure
+		}
+		defer adminClient.Close()
+		defer client.Close()
+		if !sourceProfile.UseTargetSchema() {
+			err = validateExistingDb(ctx, conv.TargetDb, dbURI, adminClient, client, conv)
+			if err != nil {
+				err = fmt.Errorf("error while validating existing database: %v", err)
+				return subcommands.ExitFailure
+			}
+		}
+		bw, err = conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
+		if err != nil {
+			err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 			return subcommands.ExitFailure
 		}
-	}
 
-	dataCoversionStartTime := time.Now()
-	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
-	if err != nil {
-		err = fmt.Errorf("can't finish data migration: %v", err)
-		return subcommands.ExitFailure
+		if !cmd.skipForeignKeys {
+			if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
+				err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
+				return subcommands.ExitFailure
+			}
+		}
+		banner = utils.GetBanner(now, dbURI)
+	} else {
+		conv.Audit.DryRun = true
+		bw, err = conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, nil, conv, true, cmd.writeLimit)
+		if err != nil {
+			err = fmt.Errorf("can't finish data conversion for db %s: %v", dbName, err)
+			return subcommands.ExitFailure
+		}
+		banner = utils.GetBanner(dataCoversionStartTime, dbName)
 	}
 	dataCoversionEndTime := time.Now()
 	dataCoversionDuration := dataCoversionEndTime.Sub(dataCoversionStartTime)
 	conv.Audit.DataConversionDuration = dataCoversionDuration
 
-	if !cmd.skipForeignKeys {
-		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
-			err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
-			return subcommands.ExitFailure
-		}
-	}
-
-	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
 	// Cleanup hb tmp data directory.
 	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	return subcommands.ExitSuccess
 }
+
+// validateExistingDb validates that the existing spanner schema is in accordance with the one specified in the session file.
+func validateExistingDb(ctx context.Context, targetDb, dbURI string, adminClient *database.DatabaseAdminClient, client *sp.Client, conv *internal.Conv) error {
+	dbExists, err := conversion.CheckExistingDb(ctx, adminClient, dbURI)
+	if err != nil {
+		err = fmt.Errorf("can't verify target database: %v", err)
+		return err
+	}
+	if !dbExists {
+		err = fmt.Errorf("target database doesn't exist")
+		return err
+	}
+	err = conversion.ValidateTables(ctx, client, targetDb)
+	if err != nil {
+		err = fmt.Errorf("error validating the tables: %v", err)
+		return err
+	}
+	spannerConv := internal.MakeConv()
+	spannerConv.TargetDb = targetDb
+	err = utils.ReadSpannerSchema(ctx, spannerConv, client)
+	if err != nil {
+		err = fmt.Errorf("can't read spanner schema: %v", err)
+		return err
+	}
+	err = utils.CompareSchema(conv, spannerConv)
+	if err != nil {
+		err = fmt.Errorf("error while comparing the schema from session file and existing spanner schema: %v", err)
+		return err
+	}
+	return nil
+}
diff --git a/cmd/schema.go b/cmd/schema.go
index 3afd93c..a62a79d 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -22,13 +22,15 @@ import (
 	"path"
 	"time"
 
+	database "cloud.google.com/go/spanner/admin/database/apiv1"
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/logger"
-	"github.com/cloudspannerecosystem/harbourbridge/profiles"
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/google/subcommands"
+	"github.com/google/uuid"
 	"go.uber.org/zap"
 )
 
@@ -40,6 +42,7 @@ type SchemaCmd struct {
 	targetProfile string
 	filePrefix    string // TODO: move filePrefix to global flags
 	logLevel      string
+	dryRun        bool
 }
 
 // Name returns the name of operation.
@@ -71,6 +74,7 @@ func (cmd *SchemaCmd) SetFlags(f *flag.FlagSet) {
 	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.BoolVar(&cmd.dryRun, "dry-run", false, "Flag for generating DDL and schema conversion report without creating a spanner database")
 }
 
 func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus {
@@ -89,37 +93,14 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 	}
 	defer logger.Log.Sync()
 
-	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
+	sourceProfile, targetProfile, ioHelper, dbName, err := PrepareMigrationPrerequisites(cmd.sourceProfile, cmd.targetProfile, cmd.source)
 	if err != nil {
+		err = fmt.Errorf("error while preparing prerequisites for migration: %v", err)
 		return subcommands.ExitUsageError
 	}
-	sourceProfile.Driver, err = sourceProfile.ToLegacyDriver(cmd.source)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-
-	targetProfile, err := profiles.NewTargetProfile(cmd.targetProfile)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-	targetProfile.TargetDb = targetProfile.ToLegacyTargetDb()
-
-	dumpFilePath := ""
-	if sourceProfile.Ty == profiles.SourceProfileTypeFile && (sourceProfile.File.Format == "" || sourceProfile.File.Format == "dump") {
-		dumpFilePath = sourceProfile.File.Path
-	}
-	ioHelper := utils.NewIOStreams(sourceProfile.Driver, dumpFilePath)
-	if ioHelper.SeekableIn != nil {
-		defer ioHelper.In.Close()
-	}
 
 	// If filePrefix not explicitly set, use generated dbName.
 	if cmd.filePrefix == "" {
-		dbName, err := utils.GetDatabaseName(sourceProfile.Driver, time.Now())
-		if err != nil {
-			err = fmt.Errorf("can't generate database name for prefix: %v", err)
-			return subcommands.ExitFailure
-		}
 		cmd.filePrefix = dbName + "."
 	}
 
@@ -130,11 +111,37 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 		return subcommands.ExitFailure
 	}
 
+	conversion.WriteSchemaFile(conv, schemaConversionStartTime, cmd.filePrefix+schemaFile, ioHelper.Out)
+	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
+
+	// Populate migration request id and migration type in conv object.
+	conv.Audit.MigrationRequestId = "HB-" + uuid.New().String()
+	conv.Audit.MigrationType = migration.MigrationData_SCHEMA_ONLY.Enum()
+
+	var (
+		adminClient *database.DatabaseAdminClient
+		dbURI       string
+	)
+
+	if !cmd.dryRun {
+		adminClient, _, dbURI, err = CreateDatabaseClient(ctx, targetProfile, sourceProfile.Driver, ioHelper)
+		if err != nil {
+			err = fmt.Errorf("can't create database client: %v", err)
+			return subcommands.ExitFailure
+		}
+		defer adminClient.Close()
+
+		err = conversion.CreateOrUpdateDatabase(ctx, adminClient, dbURI, sourceProfile.Driver, targetProfile.TargetDb, conv, ioHelper.Out)
+		if err != nil {
+			err = fmt.Errorf("can't create/update database: %v", err)
+			return subcommands.ExitFailure
+		}
+	}
+
 	schemaCoversionEndTime := time.Now()
 	conv.Audit.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
-	conversion.WriteSchemaFile(conv, schemaCoversionEndTime, cmd.filePrefix+schemaFile, ioHelper.Out)
-	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
-	conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix+reportFile, ioHelper.Out)
+	banner := utils.GetBanner(schemaConversionStartTime, dbName)
+	conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	// Cleanup hb tmp data directory.
 	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	return subcommands.ExitSuccess
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index 282e2df..c53a1f9 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -22,13 +22,15 @@ import (
 	"path"
 	"time"
 
+	sp "cloud.google.com/go/spanner"
+	database "cloud.google.com/go/spanner/admin/database/apiv1"
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/logger"
-	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
+	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
 	"github.com/google/subcommands"
 	"github.com/google/uuid"
 	"go.uber.org/zap"
@@ -77,7 +79,7 @@ func (cmd *SchemaAndDataCmd) SetFlags(f *flag.FlagSet) {
 	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)")
 	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, "To validate the syntax of the command by running it in an air-gapped manner, such that no network calls are made.")
+	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")
 }
 
@@ -90,10 +92,6 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 			logger.Log.Fatal("FATAL error", zap.Error(err))
 		}
 	}()
-	if cmd.dryRun {
-		fmt.Print("--dry-run flag is not implemented")
-		return subcommands.ExitFailure
-	}
 	err = logger.InitializeLogger(cmd.logLevel)
 	if err != nil {
 		fmt.Println("Error initialising logger, did you specify a valid log-level? [DEBUG, INFO, WARN, ERROR, FATAL]", err)
@@ -101,42 +99,26 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	}
 	defer logger.Log.Sync()
 
-	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-	sourceProfile.Driver, err = sourceProfile.ToLegacyDriver(cmd.source)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-
-	targetProfile, err := profiles.NewTargetProfile(cmd.targetProfile)
+	sourceProfile, targetProfile, ioHelper, dbName, err := PrepareMigrationPrerequisites(cmd.sourceProfile, cmd.targetProfile, cmd.source)
 	if err != nil {
+		err = fmt.Errorf("error while preparing prerequisites for migration: %v", err)
 		return subcommands.ExitUsageError
 	}
-	targetProfile.TargetDb = targetProfile.ToLegacyTargetDb()
-
-	dumpFilePath := ""
-	if sourceProfile.Ty == profiles.SourceProfileTypeFile && (sourceProfile.File.Format == "" || sourceProfile.File.Format == "dump") {
-		dumpFilePath = sourceProfile.File.Path
-	}
-	ioHelper := utils.NewIOStreams(sourceProfile.Driver, dumpFilePath)
-	if ioHelper.SeekableIn != nil {
-		defer ioHelper.In.Close()
-	}
-
 	schemaConversionStartTime := time.Now()
 
 	// If filePrefix not explicitly set, use dbName as prefix.
 	if cmd.filePrefix == "" {
-		dbName, err := utils.GetDatabaseName(sourceProfile.Driver, schemaConversionStartTime)
-		if err != nil {
-			panic(fmt.Errorf("can't generate database name for prefix: %v", err))
-		}
 		cmd.filePrefix = dbName + "."
 	}
 
-	var conv *internal.Conv
+	var (
+		conv        *internal.Conv
+		bw          *writer.BatchWriter
+		banner      string
+		dbURI       string
+		adminClient *database.DatabaseAdminClient
+		client      *sp.Client
+	)
 	conv, err = conversion.SchemaConv(sourceProfile, targetProfile, &ioHelper)
 	if err != nil {
 		panic(err)
@@ -144,62 +126,62 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	schemaCoversionEndTime := time.Now()
 	conv.Audit.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
 
-	// Populate migration request id and migration type in conv object
+	// Populate migration request id and migration type in conv object.
 	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.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
-	conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix+reportFile, ioHelper.Out)
 
-	project, instance, dbName, err := targetProfile.GetResourceIds(ctx, schemaConversionStartTime, sourceProfile.Driver, ioHelper.Out)
-	if err != nil {
-		return subcommands.ExitUsageError
-	}
-	fmt.Println("Using Google Cloud project:", project)
-	fmt.Println("Using Cloud Spanner instance:", instance)
-	utils.PrintPermissionsWarning(sourceProfile.Driver, ioHelper.Out)
-
-	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName)
-
-	adminClient, err := utils.NewDatabaseAdminClient(ctx)
-	if err != nil {
-		err = fmt.Errorf("can't create admin client: %w", utils.AnalyzeError(err, dbURI))
-		return subcommands.ExitFailure
-	}
-	defer adminClient.Close()
-	client, err := utils.GetClient(ctx, dbURI)
-	if err != nil {
-		err = fmt.Errorf("can't create client for db %s: %v", dbURI, err)
-		return subcommands.ExitFailure
-	}
-	defer client.Close()
-
-	err = conversion.CreateOrUpdateDatabase(ctx, adminClient, dbURI, sourceProfile.Driver, targetProfile.TargetDb, conv, ioHelper.Out)
-	if err != nil {
-		err = fmt.Errorf("can't create/update database: %v", err)
-		return subcommands.ExitFailure
-	}
+	if !cmd.dryRun {
+		conversion.Report(sourceProfile.Driver, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix+reportFile, ioHelper.Out)
+		adminClient, client, dbURI, err = CreateDatabaseClient(ctx, targetProfile, sourceProfile.Driver, ioHelper)
+		if err != nil {
+			err = fmt.Errorf("can't create database client: %v", err)
+			return subcommands.ExitFailure
+		}
+		defer adminClient.Close()
+		defer client.Close()
 
-	dataCoversionStartTime := time.Now()
-	bw, err := conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
-	if err != nil {
-		err = fmt.Errorf("can't finish data migration: %v", err)
-		return subcommands.ExitFailure
-	}
-	dataCoversionEndTime := time.Now()
-	dataCoversionDuration := dataCoversionEndTime.Sub(dataCoversionStartTime)
-	conv.Audit.DataConversionDuration = dataCoversionDuration
+		err = conversion.CreateOrUpdateDatabase(ctx, adminClient, dbURI, sourceProfile.Driver, targetProfile.TargetDb, conv, ioHelper.Out)
+		if err != nil {
+			err = fmt.Errorf("can't create/update database: %v", err)
+			return subcommands.ExitFailure
+		}
+		schemaCoversionEndTime := time.Now()
+		conv.Audit.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
 
-	if !cmd.skipForeignKeys {
-		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
-			err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
+		bw, err = conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, client, conv, true, cmd.writeLimit)
+		if err != nil {
+			err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 			return subcommands.ExitFailure
 		}
+		if !cmd.skipForeignKeys {
+			if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
+				err = fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
+				return subcommands.ExitFailure
+			}
+		}
+		dataCoversionEndTime := time.Now()
+		conv.Audit.DataConversionDuration = dataCoversionEndTime.Sub(schemaCoversionEndTime)
+		banner = utils.GetBanner(schemaConversionStartTime, dbURI)
+
+	} else {
+		conv.Audit.DryRun = true
+		schemaCoversionEndTime := time.Now()
+		conv.Audit.SchemaConversionDuration = schemaCoversionEndTime.Sub(schemaConversionStartTime)
+		bw, err = conversion.DataConv(ctx, sourceProfile, targetProfile, &ioHelper, nil, conv, true, cmd.writeLimit)
+		if err != nil {
+			err = fmt.Errorf("can't finish data conversion for db %s: %v", dbName, err)
+			return subcommands.ExitFailure
+		}
+		dataCoversionEndTime := time.Now()
+		conv.Audit.DataConversionDuration = dataCoversionEndTime.Sub(schemaCoversionEndTime)
+		banner = utils.GetBanner(schemaConversionStartTime, dbName)
 	}
-	banner := utils.GetBanner(schemaConversionStartTime, dbURI)
 	conversion.Report(sourceProfile.Driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
+
 	// Cleanup hb tmp data directory.
 	os.RemoveAll(os.TempDir() + constants.HB_TMP_DIR)
 	return subcommands.ExitSuccess
diff --git a/cmd/utils.go b/cmd/utils.go
new file mode 100644
index 0000000..deaff58
--- /dev/null
+++ b/cmd/utils.go
@@ -0,0 +1,84 @@
+// Copyright 2022 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 cmd
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	sp "cloud.google.com/go/spanner"
+	database "cloud.google.com/go/spanner/admin/database/apiv1"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
+)
+
+// CreateDatabaseClient creates new database client and admin client.
+func CreateDatabaseClient(ctx context.Context, targetProfile profiles.TargetProfile, driver string, ioHelper utils.IOStreams) (*database.DatabaseAdminClient, *sp.Client, string, error) {
+	project, instance, dbName, err := targetProfile.GetResourceIds(ctx, time.Now(), driver, ioHelper.Out)
+	if err != nil {
+		return nil, nil, "", err
+	}
+	fmt.Println("Using Google Cloud project:", project)
+	fmt.Println("Using Cloud Spanner instance:", instance)
+	utils.PrintPermissionsWarning(driver, ioHelper.Out)
+
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName)
+	adminClient, err := utils.NewDatabaseAdminClient(ctx)
+	if err != nil {
+		err = fmt.Errorf("can't create admin client: %v", utils.AnalyzeError(err, dbURI))
+		return nil, nil, dbURI, err
+	}
+	client, err := utils.GetClient(ctx, dbURI)
+	if err != nil {
+		err = fmt.Errorf("can't create client for db %s: %v", dbURI, err)
+		return adminClient, nil, dbURI, err
+	}
+	return adminClient, client, dbURI, nil
+}
+
+// PrepareMigrationPrerequisites creates source and target profiles, opens a new IOStream and generates the database name.
+func PrepareMigrationPrerequisites(sourceProfileString, targetProfileString, source string) (profiles.SourceProfile, profiles.TargetProfile, utils.IOStreams, string, error) {
+	sourceProfile, err := profiles.NewSourceProfile(sourceProfileString, source)
+	if err != nil {
+		return profiles.SourceProfile{}, profiles.TargetProfile{}, utils.IOStreams{}, "", err
+	}
+	sourceProfile.Driver, err = sourceProfile.ToLegacyDriver(source)
+	if err != nil {
+		return profiles.SourceProfile{}, profiles.TargetProfile{}, utils.IOStreams{}, "", err
+	}
+
+	targetProfile, err := profiles.NewTargetProfile(targetProfileString)
+	if err != nil {
+		return sourceProfile, profiles.TargetProfile{}, utils.IOStreams{}, "", err
+	}
+	targetProfile.TargetDb = targetProfile.ToLegacyTargetDb()
+
+	dumpFilePath := ""
+	if sourceProfile.Ty == profiles.SourceProfileTypeFile && (sourceProfile.File.Format == "" || sourceProfile.File.Format == "dump") {
+		dumpFilePath = sourceProfile.File.Path
+	}
+	ioHelper := utils.NewIOStreams(sourceProfile.Driver, dumpFilePath)
+	if ioHelper.SeekableIn != nil {
+		defer ioHelper.In.Close()
+	}
+
+	dbName, err := utils.GetDatabaseName(sourceProfile.Driver, time.Now())
+	if err != nil {
+		err = fmt.Errorf("can't generate database name for prefix: %v", err)
+		return sourceProfile, targetProfile, ioHelper, "", err
+	}
+	return sourceProfile, targetProfile, ioHelper, dbName, nil
+}
diff --git a/common/utils/utils.go b/common/utils/utils.go
index 139ae8c..c69d9df 100644
--- a/common/utils/utils.go
+++ b/common/utils/utils.go
@@ -506,6 +506,58 @@ func ReadSpannerSchema(ctx context.Context, conv *internal.Conv, client *sp.Clie
 	return nil
 }
 
+// CompareSchema compares the spanner schema of two conv objects and returns specific error if they don't match
+func CompareSchema(conv1, conv2 *internal.Conv) error {
+	if conv1.TargetDb != conv2.TargetDb {
+		return fmt.Errorf("target db don't match")
+	}
+	for spannerTableInd := range conv1.SpSchema {
+		sessionTable := conv1.SpSchema[spannerTableInd]
+		spannerTable := conv2.SpSchema[spannerTableInd]
+		if sessionTable.Name != spannerTable.Name || sessionTable.Parent != spannerTable.Parent ||
+			len(sessionTable.Pks) != len(spannerTable.Pks) || len(sessionTable.ColDefs) != len(spannerTable.ColDefs) ||
+			len(sessionTable.Indexes) != len(spannerTable.Indexes) {
+			return fmt.Errorf("table detail for table %v don't match", sessionTable.Name)
+		}
+		for primaryKeyIndex := range sessionTable.Pks {
+			if sessionTable.Pks[primaryKeyIndex].Col != spannerTable.Pks[primaryKeyIndex].Col || sessionTable.Pks[primaryKeyIndex].Desc != spannerTable.Pks[primaryKeyIndex].Desc {
+				return fmt.Errorf("primary keys for table %v don't match", sessionTable.Name)
+			}
+		}
+		for col := range sessionTable.ColDefs {
+			colDef := sessionTable.ColDefs[col]
+			spannerCol := spannerTable.ColDefs[col]
+			if colDef.Name != spannerCol.Name || colDef.NotNull != spannerCol.NotNull ||
+				colDef.T.IsArray != spannerCol.T.IsArray || colDef.T.Len != spannerCol.T.Len || colDef.T.Name != spannerCol.T.Name {
+				return fmt.Errorf("column detail for table %v don't match", sessionTable.Name)
+			}
+		}
+		for _, sessionTableIndex := range sessionTable.Indexes {
+			found := 0
+			for _, spannerTableIndex := range spannerTable.Indexes {
+				if sessionTableIndex.Name == spannerTableIndex.Name {
+					found = 1
+					if sessionTableIndex.Table != spannerTableIndex.Table || sessionTableIndex.Unique != spannerTableIndex.Unique ||
+						len(sessionTableIndex.Keys) != len(spannerTableIndex.Keys) {
+						return fmt.Errorf("index %v - details don't match", sessionTableIndex.Name)
+					}
+					for keyIndex := range sessionTableIndex.Keys {
+						if sessionTableIndex.Keys[keyIndex].Col != spannerTableIndex.Keys[keyIndex].Col ||
+							sessionTableIndex.Keys[keyIndex].Desc != spannerTableIndex.Keys[keyIndex].Desc {
+							return fmt.Errorf("index %v - keys don't match", sessionTableIndex.Name)
+						}
+					}
+					break
+				}
+			}
+			if found == 0 {
+				return fmt.Errorf("index %v not found in spanner schema", sessionTableIndex.Name)
+			}
+		}
+	}
+	return nil
+}
+
 func DialectToTarget(dialect string) string {
 	if strings.ToLower(dialect) == constants.DIALECT_POSTGRESQL {
 		return constants.TargetExperimentalPostgres
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 731c0ea..2156c98 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -56,6 +56,7 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/sources/mysql"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/oracle"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/postgres"
+	"github.com/cloudspannerecosystem/harbourbridge/sources/spanner"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/sqlserver"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/writer"
@@ -176,7 +177,10 @@ func schemaFromDatabase(sourceProfile profiles.SourceProfile, targetProfile prof
 func performSnapshotMigration(config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client, infoSchema common.InfoSchema) (*writer.BatchWriter, error) {
 	common.SetRowStats(conv, infoSchema)
 	totalRows := conv.Rows()
-	p := internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
+	var p *internal.Progress
+	if !conv.Audit.DryRun {
+		p = internal.NewProgress(totalRows, "Writing data to Spanner", internal.Verbose(), false)
+	}
 	batchWriter := populateDataConv(conv, config, client, p)
 	common.ProcessData(conv, infoSchema)
 	batchWriter.Flush()
@@ -337,13 +341,16 @@ func populateDataConv(conv *internal.Conv, config writer.BatchWriterConfig, clie
 	}
 	batchWriter := writer.NewBatchWriter(config)
 	conv.SetDataMode()
-	conv.SetDataSink(
-		func(table string, cols []string, vals []interface{}) {
-			batchWriter.AddRow(table, cols, vals)
-		})
-	conv.DataFlush = func() {
-		batchWriter.Flush()
+	if !conv.Audit.DryRun {
+		conv.SetDataSink(
+			func(table string, cols []string, vals []interface{}) {
+				batchWriter.AddRow(table, cols, vals)
+			})
+		conv.DataFlush = func() {
+			batchWriter.Flush()
+		}
 	}
+
 	return batchWriter
 }
 
@@ -450,6 +457,25 @@ func CheckExistingDb(ctx context.Context, adminClient *database.DatabaseAdminCli
 	}
 }
 
+// ValidateTables validates that all the tables in the database are empty.
+func ValidateTables(ctx context.Context, client *sp.Client, targetDb string) error {
+	infoSchema := spanner.InfoSchemaImpl{Client: client, Ctx: ctx, TargetDb: targetDb}
+	tables, err := infoSchema.GetTables()
+	if err != nil {
+		return err
+	}
+	for _, table := range tables {
+		count, err := infoSchema.GetRowCount(table)
+		if err != nil {
+			return err
+		}
+		if count != 0 {
+			return fmt.Errorf("table %v should be empty for data migration to take place", table.Name)
+		}
+	}
+	return nil
+}
+
 // ValidateDDL verifies if an existing DB's ddl follows what is supported by harbourbridge. Currently,
 // we only support empty schema when db already exists.
 func ValidateDDL(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string) error {
diff --git a/internal/convert.go b/internal/convert.go
index 1c2953d..c62b766 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -43,7 +43,7 @@ type Conv struct {
 	TimezoneOffset string              // Timezone offset for timestamp conversion.
 	TargetDb       string              // The target database to which HarbourBridge is writing.
 	UniquePKey     map[string][]string // Maps Spanner table name to unique column name being used as primary key (if needed).
-	Audit          audit               // Stores the audit information for the database conversion
+	Audit          Audit               // Stores the audit information for the database conversion
 }
 
 type mode int
@@ -141,14 +141,15 @@ type statementStat struct {
 }
 
 // Stores the audit information of conversion.
-// Elements that do not affect the migration functionality but are relevant for the conversion report.
-type audit struct {
+// Elements that do not affect the migration functionality but are relevant for the migration metadata.
+type Audit struct {
 	ToSpannerFkIdx           map[string]FkeyAndIdxs                 `json:"-"` // Maps from source-DB table name to Spanner names for table name, foreign key and indexes.
 	ToSourceFkIdx            map[string]FkeyAndIdxs                 `json:"-"` // Maps from Spanner table name to source-DB names for table name, foreign key and indexes.
 	SchemaConversionDuration time.Duration                          `json:"-"` // Duration of schema conversion.
 	DataConversionDuration   time.Duration                          `json:"-"` // Duration of data conversion.
 	MigrationRequestId       string                                 `json:"-"` // Unique request id generated per migration
 	MigrationType            *migration.MigrationData_MigrationType `json:"-"` // Type of migration: Schema migration, data migration or schema and data migration
+	DryRun                   bool                                   `json:"-"` // Flag to identify if the migration is a dry run.
 	StreamingStats           streamingStats                         `json:"-"` // Stores information related to streaming migration process.
 }
 
@@ -183,10 +184,11 @@ func MakeConv() *Conv {
 		},
 		TimezoneOffset: "+00:00", // By default, use +00:00 offset which is equal to UTC timezone
 		UniquePKey:     make(map[string][]string),
-		Audit: audit{
+		Audit: Audit{
 			ToSpannerFkIdx: make(map[string]FkeyAndIdxs),
 			ToSourceFkIdx:  make(map[string]FkeyAndIdxs),
 			StreamingStats: streamingStats{},
+			MigrationType:  migration.MigrationData_SCHEMA_ONLY.Enum(),
 		},
 	}
 }
@@ -220,7 +222,9 @@ func (conv *Conv) SetDataMode() {
 
 // WriteRow calls dataSink and updates row stats.
 func (conv *Conv) WriteRow(srcTable, spTable string, spCols []string, spVals []interface{}) {
-	if conv.dataSink == nil {
+	if conv.Audit.DryRun {
+		conv.statsAddGoodRow(srcTable, conv.DataMode())
+	} else if conv.dataSink == nil {
 		msg := "Internal error: ProcessDataRow called but dataSink not configured"
 		VerbosePrintf("%s\n", msg)
 		logger.Log.Debug("Internal error: ProcessDataRow called but dataSink not configured")
diff --git a/internal/report.go b/internal/report.go
index fa1d5b0..215445a 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -22,7 +22,7 @@ import (
 	"strings"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 )
@@ -43,6 +43,14 @@ func GenerateReport(driverName string, conv *Conv, w *bufio.Writer, badWrites ma
 			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") {
@@ -52,8 +60,8 @@ func GenerateReport(driverName string, conv *Conv, w *bufio.Writer, badWrites ma
 		statementsMsg = "stats on the " + driverName + " statements processed, followed by "
 	}
 	justifyLines(w, "The remainder of this report provides "+statementsMsg+
-		"a table-by-table listing of schema and data conversion details. "+
-		"For background on the schema and data conversion process used, "+
+		"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")
@@ -69,7 +77,7 @@ func GenerateReport(driverName string, conv *Conv, w *bufio.Writer, badWrites ma
 				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()))
+			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)
@@ -254,16 +262,19 @@ func buildTableReport(conv *Conv, srcTable string, badWrites map[string]int64) t
 		tr.Body = []tableReportBody{{Heading: "Internal error: " + m}}
 		return tr
 	}
-	issues, cols, warnings := AnalyzeCols(conv, srcTable, spTable)
-	tr.Cols = cols
-	tr.Warnings = warnings
-	if pk, ok := conv.SyntheticPKeys[spTable]; ok {
-		tr.SyntheticPKey = pk.Col
-		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, &pk.Col, nil)
-	} else if pk, ok := conv.UniquePKey[spTable]; ok {
-		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil, pk)
-	} else {
-		tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil, nil)
+	if *conv.Audit.MigrationType != migration.MigrationData_DATA_ONLY {
+		issues, cols, warnings := AnalyzeCols(conv, srcTable, spTable)
+		tr.Cols = cols
+		tr.Warnings = warnings
+		if pk, ok := conv.SyntheticPKeys[spTable]; ok {
+			tr.SyntheticPKey = pk.Col
+			tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, &pk.Col, nil)
+		} else if pk, ok := conv.UniquePKey[spTable]; ok {
+			tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil, pk)
+		} else {
+			tr.Body = buildTableReportBody(conv, srcTable, issues, spSchema, srcSchema, nil, nil)
+		}
+
 	}
 	if !conv.SchemaMode() {
 		fillRowStats(conv, srcTable, badWrites, &tr)
@@ -542,13 +553,19 @@ func rateSchema(cols, warnings int64, missingPKey, summary bool) string {
 	}
 }
 
-func rateData(rows int64, badRows int64) string {
-	s := fmt.Sprintf(" (%s%% of %d rows written to Spanner)", pct(rows, badRows), rows)
+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 written to Spanner)", rows)
+		return fmt.Sprintf("EXCELLENT (all %d rows %s to Spanner)", rows, reportText)
 	case good(rows, badRows):
 		return "GOOD" + s
 	case ok(rows, badRows):
@@ -566,10 +583,13 @@ func ok(total, badCount int64) bool {
 	return float64(badCount) < float64(total)/3
 }
 
-func rateConversion(rows, badRows, cols, warnings int64, missingPKey, summary bool, schemaOnly bool) string {
-	rate := fmt.Sprintf("Schema conversion: %s.\n", rateSchema(cols, warnings, missingPKey, summary))
+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 {
+		rate = rate + fmt.Sprintf("Schema conversion: %s.\n", rateSchema(cols, warnings, missingPKey, summary))
+	}
 	if !schemaOnly {
-		rate = rate + fmt.Sprintf("Data conversion: %s.\n", rateData(rows, badRows))
+		rate = rate + fmt.Sprintf("Data conversion: %s.\n", rateData(rows, badRows, dryRun))
 	}
 	return rate
 }
@@ -600,7 +620,7 @@ func GenerateSummary(conv *Conv, r []tableReport, badWrites map[string]int64) st
 	for _, n := range badWrites {
 		badRows += n
 	}
-	return rateConversion(rows, badRows, cols, warnings, missingPKey, true, conv.SchemaMode())
+	return rateConversion(rows, badRows, cols, warnings, missingPKey, true, conv.SchemaMode(), *conv.Audit.MigrationType, conv.Audit.DryRun)
 }
 
 // IgnoredStatements creates a list of statements to ignore.
diff --git a/sources/dynamodb/streaming.go b/sources/dynamodb/streaming.go
index a22aa45..db1e24a 100644
--- a/sources/dynamodb/streaming.go
+++ b/sources/dynamodb/streaming.go
@@ -153,44 +153,64 @@ func ProcessStream(wgStream *sync.WaitGroup, streamClient dynamodbstreamsiface.D
 	defer wgStream.Done()
 	wgShard := &sync.WaitGroup{}
 
-	var lastProcessedShardId *string = nil
+	processingStarted := make(map[string]bool)
+
 	passAfterUserExit := false
 	for {
-		result, err := fetchShards(streamClient, lastProcessedShardId, streamArn)
+		shards, err := scanShards(streamClient, streamArn)
 		if err != nil {
-			streamInfo.Unexpected(fmt.Sprintf("Couldn't fetch shards for table %s: %s", srcTable, err))
+			streamInfo.Unexpected(fmt.Sprintf("Couldn't scan shards for table %s: %s", srcTable, err))
 			break
 		}
-		for _, shard := range result.Shards {
-			lastProcessedShardId = shard.ShardId
+		for _, shard := range shards {
+			shardId := *shard.ShardId
+			if _, ok := processingStarted[shardId]; !ok {
+				processingStarted[shardId] = false
+				streamInfo.SetShardStatus(shardId, false)
+			}
+		}
+		for _, shard := range shards {
+			shardId := *shard.ShardId
+			if !processingStarted[shardId] {
+				processingStarted[shardId] = true
 
-			wgShard.Add(1)
-			go ProcessShard(wgShard, streamInfo, conv, streamClient, shard, streamArn, srcTable)
+				wgShard.Add(1)
+				go ProcessShard(wgShard, streamInfo, conv, streamClient, shard, streamArn, srcTable)
+			}
 		}
 
-		if result.LastEvaluatedShardId == nil && passAfterUserExit {
+		if passAfterUserExit {
 			break
-		}
-		if streamInfo.UserExit {
+		} else if streamInfo.UserExit {
 			passAfterUserExit = true
-		} else if len(result.Shards) == 0 {
-			time.Sleep(10 * time.Second)
+		} else {
+			time.Sleep(20 * time.Second)
 		}
 	}
 	wgShard.Wait()
 }
 
-// fetchShards fetches latest unprocessed shards from a given DynamoDB Stream after the lastProcessedShardId.
-func fetchShards(streamClient dynamodbstreamsiface.DynamoDBStreamsAPI, lastProcessedShardId *string, streamArn string) (*dynamodbstreams.StreamDescription, error) {
+// scanShards fetches all the shards from a given DynamoDB Stream.
+func scanShards(streamClient dynamodbstreamsiface.DynamoDBStreamsAPI, streamArn string) ([]*dynamodbstreams.Shard, error) {
 	describeStreamInput := &dynamodbstreams.DescribeStreamInput{
-		ExclusiveStartShardId: lastProcessedShardId,
+		ExclusiveStartShardId: nil,
 		StreamArn:             &streamArn,
 	}
-	result, err := streamClient.DescribeStream(describeStreamInput)
-	if err != nil {
-		return nil, fmt.Errorf("unexpected call to DescribeStream: %v", err)
+	var scanResult []*dynamodbstreams.Shard
+	for {
+		result, err := streamClient.DescribeStream(describeStreamInput)
+		if err != nil {
+			return nil, fmt.Errorf("unexpected call to DescribeStream: %v", err)
+		}
+		scanResult = append(scanResult, result.StreamDescription.Shards...)
+
+		if result.StreamDescription.LastEvaluatedShardId == nil {
+			break
+		} else {
+			describeStreamInput.ExclusiveStartShardId = result.StreamDescription.LastEvaluatedShardId
+		}
 	}
-	return result.StreamDescription, nil
+	return scanResult, nil
 }
 
 // checkTrimmedDataError checks if the error is an TrimmedDataAccessException.
@@ -208,7 +228,6 @@ func ProcessShard(wgShard *sync.WaitGroup, streamInfo *StreamingInfo, conv *inte
 	waitForParentShard(streamInfo, shard.ParentShardId)
 
 	shardId := *shard.ShardId
-	streamInfo.SetShardStatus(shardId, false)
 
 	var lastEvaluatedSequenceNumber *string = nil
 	passAfterUserExit := false
@@ -267,9 +286,9 @@ func waitForParentShard(streamInfo *StreamingInfo, parentShard *string) {
 	if parentShard != nil {
 		for {
 			streamInfo.lock.Lock()
-			done := streamInfo.ShardProcessed[*parentShard]
+			done, ok := streamInfo.ShardProcessed[*parentShard]
 			streamInfo.lock.Unlock()
-			if done {
+			if !ok || done {
 				return
 			} else {
 				time.Sleep(6 * time.Second)
diff --git a/sources/spanner/infoschema.go b/sources/spanner/infoschema.go
index df47ae6..e490a8b 100644
--- a/sources/spanner/infoschema.go
+++ b/sources/spanner/infoschema.go
@@ -50,8 +50,25 @@ func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcS
 	return nil
 }
 
+// GetRowCount returns the row count of the table.
 func (isi InfoSchemaImpl) GetRowCount(table common.SchemaAndName) (int64, error) {
-	return 0, nil
+	q := "SELECT count(*) FROM " + table.Name + ";"
+	stmt := spanner.Statement{
+		SQL: q,
+	}
+	iter := isi.Client.Single().Query(isi.Ctx, stmt)
+	defer iter.Stop()
+	var count int64
+	row, err := iter.Next()
+	if err == iterator.Done {
+		return 0, nil
+	}
+	if err != nil {
+		return count, err
+	}
+	row.Columns(&count)
+	return count, err
+
 }
 
 func (isi InfoSchemaImpl) GetRowsFromTable(conv *internal.Conv, srcTable string) (interface{}, error) {
@@ -224,14 +241,14 @@ func (isi InfoSchemaImpl) GetConstraints(conv *internal.Conv, table common.Schem
 
 // GetForeignKeys returns a list of all the foreign key constraints.
 func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.SchemaAndName) (foreignKeys []schema.ForeignKey, err error) {
-	q := `SELECT  k.constraint_name, k.column_name, t.table_name, c.column_name 
+	q := `SELECT  k.constraint_name, k.column_name, c.table_name, c.column_name 
 			FROM information_schema.key_column_usage AS k 
 			JOIN information_schema.constraint_column_usage AS c ON k.constraint_name = c.constraint_name
 			JOIN information_schema.table_constraints AS t ON k.constraint_name = t.constraint_name 
 			WHERE t.constraint_type='FOREIGN KEY' AND t.table_schema = '' AND t.table_name = @p1
 			ORDER BY k.constraint_name, k.ordinal_position;`
 	if isi.TargetDb == constants.TargetExperimentalPostgres {
-		q = `SELECT  k.constraint_name, k.column_name, t.table_name, c.column_name 
+		q = `SELECT  k.constraint_name, k.column_name, c.table_name, c.column_name 
 				FROM information_schema.key_column_usage AS k 
 				JOIN information_schema.constraint_column_usage AS c ON k.constraint_name = c.constraint_name
 				JOIN information_schema.table_constraints AS t ON k.constraint_name = t.constraint_name 
@@ -293,9 +310,60 @@ func (isi InfoSchemaImpl) GetForeignKeys(conv *internal.Conv, table common.Schem
 	return foreignKeys, nil
 }
 
-// Skipped since we dont have a use case for storing indexes yet.
+// GetIndexes returns a list of Indexes per table.
 func (isi InfoSchemaImpl) GetIndexes(conv *internal.Conv, table common.SchemaAndName) ([]schema.Index, error) {
-	return []schema.Index{}, nil
+	q := `SELECT distinct c.INDEX_NAME,c.COLUMN_NAME,c.ORDINAL_POSITION,c.COLUMN_ORDERING,i.IS_UNIQUE
+			FROM information_schema.index_columns AS c
+			JOIN information_schema.indexes AS i
+			ON c.INDEX_NAME=i.INDEX_NAME
+			WHERE c.table_schema = '' AND i.INDEX_TYPE='INDEX' AND c.TABLE_NAME = @p1 ORDER BY c.INDEX_NAME, c.ORDINAL_POSITION;`
+	if isi.TargetDb == constants.TargetExperimentalPostgres {
+		q = `SELECT distinct c.INDEX_NAME,c.COLUMN_NAME,c.ORDINAL_POSITION,c.COLUMN_ORDERING,i.IS_UNIQUE
+		FROM information_schema.index_columns AS c
+		JOIN information_schema.indexes AS i
+		ON c.INDEX_NAME=i.INDEX_NAME
+		WHERE c.table_schema = 'public' AND i.INDEX_TYPE='INDEX' AND c.TABLE_NAME = $1 ORDER BY c.INDEX_NAME, c.ORDINAL_POSITION;`
+	}
+	stmt := spanner.Statement{
+		SQL: q,
+		Params: map[string]interface{}{
+			"p1": table.Name,
+		},
+	}
+	iter := isi.Client.Single().Query(isi.Ctx, stmt)
+	defer iter.Stop()
+	var name, column, ordering string
+	var isUnique bool
+	var sequence int64
+	indexMap := make(map[string]schema.Index)
+	var indexNames []string
+	var indexes []schema.Index
+	for {
+		row, err := iter.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, fmt.Errorf("couldn't read row while fetching interleaved tables: %w", err)
+		}
+		err = row.Columns(&name, &column, &sequence, &ordering, &isUnique)
+		if err != nil {
+			fmt.Println(err)
+			conv.Unexpected(fmt.Sprintf("Can't scan: %v", err))
+			continue
+		}
+		if _, found := indexMap[name]; !found {
+			indexNames = append(indexNames, name)
+			indexMap[name] = schema.Index{Name: name, Unique: isUnique}
+		}
+		index := indexMap[name]
+		index.Keys = append(index.Keys, schema.Key{Column: column, Desc: (ordering == "DESC")})
+		indexMap[name] = index
+	}
+	for _, k := range indexNames {
+		indexes = append(indexes, indexMap[k])
+	}
+	return indexes, nil
 }
 
 func (isi InfoSchemaImpl) GetInterleaveTables() (map[string]string, error) {
diff --git a/webv2/session/remote_session_store.go b/webv2/session/remote_session_store.go
index 902d7b9..51a3cbd 100644
--- a/webv2/session/remote_session_store.go
+++ b/webv2/session/remote_session_store.go
@@ -21,6 +21,7 @@ import (
 
 	"cloud.google.com/go/spanner"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"google.golang.org/api/iterator"
 )
 
@@ -111,6 +112,9 @@ func (st *spannerStore) GetConvWithMetadata(ctx context.Context, versionId strin
 	}
 
 	convm.Conv = conv
+	convm.Conv.Audit = internal.Audit{
+		MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+	}
 	convm.SessionMetadata = SessionMetadata{
 		SessionName:  scs.SessionName,
 		EditorName:   scs.EditorName,


Test patch:
diff --git a/sources/dynamodb/streaming_test.go b/sources/dynamodb/streaming_test.go
index 6418690..38f4f28 100644
--- a/sources/dynamodb/streaming_test.go
+++ b/sources/dynamodb/streaming_test.go
@@ -130,31 +130,17 @@ func TestProcessStream(t *testing.T) {
 	assert.Equal(t, int64(1), streamInfo.TotalUnexpecteds())
 }
 
-func Test_fetchShards(t *testing.T) {
+func Test_scanShards(t *testing.T) {
 	streamArn := "testStreamArn"
 	tableName := "testTable"
 	type args struct {
-		streamClient         dynamodbstreamsiface.DynamoDBStreamsAPI
-		lastEvaluatedShardId *string
-		streamArn            string
-	}
-	streamDescription := dynamodbstreams.StreamDescription{
-		LastEvaluatedShardId: aws.String("shard2"),
-		Shards: []*dynamodbstreams.Shard{
-			{
-				ShardId: aws.String("shard1"),
-			},
-			{
-				ShardId: aws.String("shard2"),
-			},
-		},
-		StreamArn: &streamArn,
-		TableName: &tableName,
+		streamClient dynamodbstreamsiface.DynamoDBStreamsAPI
+		streamArn    string
 	}
 	tests := []struct {
 		name    string
 		args    args
-		want    *dynamodbstreams.StreamDescription
+		want    []*dynamodbstreams.Shard
 		wantErr bool
 	}{
 		{
@@ -163,13 +149,53 @@ func Test_fetchShards(t *testing.T) {
 				streamClient: &mockDynamoStreamsClient{
 					describeStreamOutputs: []dynamodbstreams.DescribeStreamOutput{
 						{
-							StreamDescription: &streamDescription,
+							StreamDescription: &dynamodbstreams.StreamDescription{
+								LastEvaluatedShardId: aws.String("shard2"),
+								Shards: []*dynamodbstreams.Shard{
+									{
+										ShardId: aws.String("shard1"),
+									},
+									{
+										ShardId: aws.String("shard2"),
+									},
+								},
+								StreamArn: &streamArn,
+								TableName: &tableName,
+							},
+						},
+						{
+							StreamDescription: &dynamodbstreams.StreamDescription{
+								LastEvaluatedShardId: nil,
+								Shards: []*dynamodbstreams.Shard{
+									{
+										ShardId: aws.String("shard3"),
+									},
+									{
+										ShardId: aws.String("shard4"),
+									},
+								},
+								StreamArn: &streamArn,
+								TableName: &tableName,
+							},
 						},
 					},
 				},
 				streamArn: streamArn,
 			},
-			want:    &streamDescription,
+			want: []*dynamodbstreams.Shard{
+				{
+					ShardId: aws.String("shard1"),
+				},
+				{
+					ShardId: aws.String("shard2"),
+				},
+				{
+					ShardId: aws.String("shard3"),
+				},
+				{
+					ShardId: aws.String("shard4"),
+				},
+			},
 			wantErr: false,
 		},
 		{
@@ -184,13 +210,13 @@ func Test_fetchShards(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := fetchShards(tt.args.streamClient, tt.args.lastEvaluatedShardId, tt.args.streamArn)
+			got, err := scanShards(tt.args.streamClient, tt.args.streamArn)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("fetchShards() error = %v, wantErr %v", err, tt.wantErr)
+				t.Errorf("scanShards() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
 			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("fetchShards() = %v, want %v", got, tt.want)
+				t.Errorf("scanShards() = %v, want %v", got, tt.want)
 			}
 		})
 	}
diff --git a/sources/mysql/report_test.go b/sources/mysql/report_test.go
index 764c29a..48aaa84 100644
--- a/sources/mysql/report_test.go
+++ b/sources/mysql/report_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/stretchr/testify/assert"
 )
@@ -60,6 +61,9 @@ func TestReport(t *testing.T) {
 	conv.Stats.BadRows = map[string]int64{"bad_schema": 10, "no_pk": 2000}
 	badWrites := map[string]int64{"bad_schema": 50, "no_pk": 0}
 	conv.Stats.Unexpected["Testing unexpected messages"] = 5
+	conv.Audit = internal.Audit{
+		MigrationType: migration.MigrationData_SCHEMA_AND_DATA.Enum(),
+	}
 	buf := new(bytes.Buffer)
 	w := bufio.NewWriter(buf)
 	internal.GenerateReport(constants.MYSQLDUMP, conv, w, badWrites, true, true)
diff --git a/sources/postgres/report_test.go b/sources/postgres/report_test.go
index 588c1c0..c0268ca 100644
--- a/sources/postgres/report_test.go
+++ b/sources/postgres/report_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/stretchr/testify/assert"
 )
@@ -56,6 +57,9 @@ func TestReport(t *testing.T) {
 	conv.Stats.BadRows = map[string]int64{"bad_schema": 10, "no_pk": 2000}
 	badWrites := map[string]int64{"bad_schema": 50, "no_pk": 0}
 	conv.Stats.Unexpected["Testing unexpected messages"] = 5
+	conv.Audit = internal.Audit{
+		MigrationType: migration.MigrationData_SCHEMA_AND_DATA.Enum(),
+	}
 	buf := new(bytes.Buffer)
 	w := bufio.NewWriter(buf)
 	internal.GenerateReport(constants.PGDUMP, conv, w, badWrites, true, true)
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index c090dd7..11e9f9b 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -183,22 +183,8 @@ func TestIntegration_MYSQL_Command(t *testing.T) {
 	checkResults(t, dbURI, true)
 }
 
-func TestIntegration_MySQLInterleaveTable_DataOnlyWithSessionFile(t *testing.T) {
-	onlyRunForEmulatorTest(t)
-	tmpdir := prepareIntegrationTest(t)
-	defer os.RemoveAll(tmpdir)
-
-	dbName := "test_interleave_table_data"
-	sessionFile := "../../test_data/mysql_interleave_session_test.json"
-
-	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
-	runDataOnlySubcommandForSessionFile(t, dbName, dbURI, sessionFile)
-	defer dropDatabase(t, dbURI)
-	checkResults(t, dbURI, true)
-}
-
 func runSchemaOnly(t *testing.T, dbName, filePrefix, sessionFile, dumpFilePath string) {
-	args := fmt.Sprintf("-driver mysqldump -schema-only -dbname %s -prefix %s < %s", dbName, filePrefix, dumpFilePath)
+	args := fmt.Sprintf("-driver mysqldump -schema-only -instance %s -dbname %s -prefix %s < %s", instanceID, dbName, filePrefix, dumpFilePath)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -214,6 +200,7 @@ func runDataOnly(t *testing.T, dbName, dbURI, filePrefix, sessionFile, dumpFileP
 }
 
 func TestIntegration_MySQLDUMP_SchemaOnly(t *testing.T) {
+	onlyRunForEmulatorTest(t)
 	tmpdir := prepareIntegrationTest(t)
 	defer os.RemoveAll(tmpdir)
 
@@ -234,6 +221,9 @@ func TestIntegration_MySQLDUMP_SchemaOnly(t *testing.T) {
 	if _, err := os.Stat(sessionFile); os.IsNotExist(err) {
 		t.Fatalf("session file not generated during schema-only test")
 	}
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	// Drop the database later.
+	defer dropDatabase(t, dbURI)
 }
 
 func TestIntegration_MySQLDUMP_DataOnly(t *testing.T) {
@@ -254,7 +244,7 @@ func TestIntegration_MySQLDUMP_DataOnly(t *testing.T) {
 }
 
 func runSchemaSubcommand(t *testing.T, dbName, filePrefix, sessionFile, dumpFilePath string) {
-	args := fmt.Sprintf("schema -prefix %s -source=mysql -target-profile='dbName=%s' < %s", filePrefix, dbName, dumpFilePath)
+	args := fmt.Sprintf("schema -prefix %s -source=mysql -target-profile='instance=%s,dbName=%s' < %s", filePrefix, instanceID, dbName, dumpFilePath)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
@@ -277,16 +267,8 @@ func runSchemaAndDataSubcommand(t *testing.T, dbName, dbURI, filePrefix, dumpFil
 	}
 }
 
-func runDataOnlySubcommandForSessionFile(t *testing.T, dbName, dbURI, sessionFile string) {
-	host, user, password := os.Getenv("MYSQLHOST"), os.Getenv("MYSQLUSER"), os.Getenv("MYSQLPWD")
-	args := fmt.Sprintf("data -source=mysql -session %s -source-profile='host=%s,user=%s,dbName=%s,password=%s' -target-profile='instance=%s,dbName=%s' ", sessionFile, host, user, dbName, password, instanceID, dbName)
-	err := common.RunCommand(args, projectID)
-	if err != nil {
-		t.Fatal(err)
-	}
-}
-
 func TestIntegration_MySQLDUMP_SchemaSubcommand(t *testing.T) {
+	onlyRunForEmulatorTest(t)
 	tmpdir := prepareIntegrationTest(t)
 	defer os.RemoveAll(tmpdir)
 
diff --git a/testing/postgres/integration_test.go b/testing/postgres/integration_test.go
index 52e8daf..505082d 100644
--- a/testing/postgres/integration_test.go
+++ b/testing/postgres/integration_test.go
@@ -154,18 +154,24 @@ func TestIntegration_PGDUMP_SchemaAndDataSubcommand(t *testing.T) {
 }
 
 func TestIntegration_PGDUMP_SchemaSubcommand(t *testing.T) {
+	onlyRunForEmulatorTest(t)
 	t.Parallel()
 
 	tmpdir := prepareIntegrationTest(t)
 	defer os.RemoveAll(tmpdir)
+	now := time.Now()
+	dbName, _ := utils.GetDatabaseName(constants.PGDUMP, now)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 
 	dataFilepath := "../../test_data/pg_dump.test.out"
 
-	args := fmt.Sprintf("schema -source=pg < %s", dataFilepath)
+	args := fmt.Sprintf("schema -source=pg -target-profile='instance=%s,dbName=%s' < %s", instanceID, dbName, dataFilepath)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
 	}
+	// Drop the database later.
+	defer dropDatabase(t, dbURI)
 }
 
 func TestIntegration_POSTGRES_Command(t *testing.T) {
@@ -221,11 +227,18 @@ func TestIntegration_POSTGRES_SchemaSubcommand(t *testing.T) {
 	tmpdir := prepareIntegrationTest(t)
 	defer os.RemoveAll(tmpdir)
 
-	args := "schema -source=postgres"
+	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+".")
+
+	args := fmt.Sprintf("schema -prefix %s -source=postgres -target-profile='instance=%s,dbName=%s'", filePrefix, instanceID, dbName)
 	err := common.RunCommand(args, projectID)
 	if err != nil {
 		t.Fatal(err)
 	}
+	// Drop the database later.
+	defer dropDatabase(t, dbURI)
 }
 
 func checkResults(t *testing.T, dbURI string) {
diff --git a/web/web_test.go b/web/web_test.go
index b0e60bf..ebc678a 100644
--- a/web/web_test.go
+++ b/web/web_test.go
@@ -25,6 +25,7 @@ import (
 
 	"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"
 	"github.com/stretchr/testify/assert"
@@ -273,6 +274,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				ToSpanner: map[string]internal.NameAndCols{
 					"t1": {Name: "t1", Cols: map[string]string{"a": "a", "b": "b", "c": "c"}},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -482,6 +486,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				ToSpanner: map[string]internal.NameAndCols{
 					"t1": {Name: "t1", Cols: map[string]string{"a": "a", "b": "b", "c": "c"}},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -544,6 +551,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				Issues: map[string]map[string][]internal.SchemaIssue{
 					"t1": {},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -605,6 +615,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				ToSource: map[string]internal.NameAndCols{
 					"t1": {Name: "t1", Cols: map[string]string{"a": "a", "b": "b", "c": "c"}},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1073,13 +1086,17 @@ func TestSetParentTable(t *testing.T) {
 				SpSchema: map[string]ddl.CreateTable{"t1": {
 					Name:     "t1",
 					ColNames: []string{"a", "b", "c"},
-					ColDefs: map[string]ddl.ColumnDef{"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
-						"b": {Name: "b", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
-						"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true}},
-					Pks: []ddl.IndexKey{{Col: "a", Desc: false}},
-					Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"a"}, ReferTable: "ref_t1", ReferColumns: []string{"ref_c1"}},
-						{Name: "fk2", Columns: []string{"c"}, ReferTable: "ref_t2", ReferColumns: []string{"ref_c2"}}},
-				}}},
+					ColDefs: map[string]ddl.ColumnDef{"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true}},
+					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a", Desc: false}},
+					Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk1", Columns: []string{"a"}, ReferTable: "ref_t1", ReferColumns: []string{"ref_c1"}},
+						ddl.Foreignkey{Name: "fk2", Columns: []string{"c"}, ReferTable: "ref_t2", ReferColumns: []string{"ref_c2"}}},
+				}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
+			},
 		},
 		{
 			name: "table with synthetic PK",
@@ -1096,7 +1113,10 @@ func TestSetParentTable(t *testing.T) {
 					Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"a"}, ReferTable: "ref_t1", ReferColumns: []string{"ref_c1"}},
 						{Name: "fk2", Columns: []string{"c"}, ReferTable: "ref_t2", ReferColumns: []string{"ref_c2"}}},
 				}},
-				SyntheticPKeys: map[string]internal.SyntheticPKey{"t1": {Col: "synth_id"}},
+				SyntheticPKeys: map[string]internal.SyntheticPKey{"t1": internal.SyntheticPKey{Col: "synth_id"}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1127,7 +1147,10 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "synth_id", Desc: false}},
 					},
 				},
-				SyntheticPKeys: map[string]internal.SyntheticPKey{"t2": {Col: "synth_id"}},
+				SyntheticPKeys: map[string]internal.SyntheticPKey{"t2": internal.SyntheticPKey{Col: "synth_id"}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1158,6 +1181,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "b", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1188,6 +1214,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1218,6 +1247,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1249,6 +1281,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "b", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1291,6 +1326,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "c", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1363,6 +1401,9 @@ func TestDropForeignKey(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1381,6 +1422,9 @@ func TestDropForeignKey(t *testing.T) {
 					"t1": {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 		{
@@ -1393,6 +1437,9 @@ func TestDropForeignKey(t *testing.T) {
 					"t1": {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 	}
@@ -1440,6 +1487,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1462,6 +1512,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx_1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx_2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1484,6 +1537,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx_1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx_2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1506,6 +1562,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx_1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx_2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1530,6 +1589,9 @@ func TestRenameIndexes(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1551,6 +1613,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1573,6 +1638,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1592,6 +1660,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1610,6 +1681,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1672,6 +1746,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1695,6 +1772,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1717,6 +1797,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1739,6 +1822,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1763,6 +1849,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1786,6 +1875,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1808,6 +1900,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1828,6 +1923,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1848,6 +1946,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1872,6 +1973,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "t2_fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "t2_fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1936,6 +2040,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1958,6 +2065,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1983,6 +2093,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2009,6 +2122,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2034,6 +2150,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2060,6 +2179,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2082,6 +2204,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2103,6 +2228,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2165,6 +2293,9 @@ func TestDropSecondaryIndex(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2183,6 +2314,9 @@ func TestDropSecondaryIndex(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 		{
@@ -2195,6 +2329,9 @@ func TestDropSecondaryIndex(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 	}
@@ -2325,6 +2462,7 @@ func buildConvMySQL(conv *internal.Conv) {
 		},
 	}
 	conv.SyntheticPKeys["t2"] = internal.SyntheticPKey{"synth_id", 0}
+	conv.Audit.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
 }
 
 func buildConvPostgres(conv *internal.Conv) {
@@ -2433,4 +2571,5 @@ func buildConvPostgres(conv *internal.Conv) {
 		},
 	}
 	conv.SyntheticPKeys["t2"] = internal.SyntheticPKey{"synth_id", 0}
+	conv.Audit.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
 }
diff --git a/webv2/primarykey/primarykey_test.go b/webv2/primarykey/primarykey_test.go
index 7523003..9f59f71 100644
--- a/webv2/primarykey/primarykey_test.go
+++ b/webv2/primarykey/primarykey_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/cloudspannerecosystem/harbourbridge/webv2/session"
 	"github.com/stretchr/testify/assert"
@@ -45,6 +46,9 @@ func TestUpdatePrimaryKey(t *testing.T) {
 				Pks: []ddl.IndexKey{{Col: "film_id", Order: 1, Desc: true}},
 				Id:  "1",
 			}},
+		Audit: internal.Audit{
+			MigrationType: migration.MigrationData_MIGRATION_TYPE_UNSPECIFIED.Enum(),
+		},
 	}
 
 	sessionState.Conv = c
@@ -112,6 +116,9 @@ func TestAddPrimaryKey(t *testing.T) {
 				Pks: []ddl.IndexKey{{Col: "film_id", Order: 1, Desc: true}},
 				Id:  "1",
 			}},
+		Audit: internal.Audit{
+			MigrationType: migration.MigrationData_MIGRATION_TYPE_UNSPECIFIED.Enum(),
+		},
 	}
 
 	sessionState.Conv = c
@@ -183,6 +190,9 @@ func TestRemovePrimaryKey(t *testing.T) {
 				Pks: []ddl.IndexKey{{Col: "film_id", Order: 1, Desc: true}, {Col: "actor_id", Order: 2, Desc: true}, {Col: "last_update", Order: 3, Desc: true}},
 				Id:  "1",
 			}},
+		Audit: internal.Audit{
+			MigrationType: migration.MigrationData_MIGRATION_TYPE_UNSPECIFIED.Enum(),
+		},
 	}
 
 	sessionState.Conv = c
@@ -253,6 +263,9 @@ func TestPrimarykey(t *testing.T) {
 				Pks: []ddl.IndexKey{{Col: "film_id", Order: 1, Desc: true}, {Col: "actor_id", Order: 2, Desc: true}},
 				Id:  "1",
 			}},
+		Audit: internal.Audit{
+			MigrationType: migration.MigrationData_MIGRATION_TYPE_UNSPECIFIED.Enum(),
+		},
 	}
 
 	sessionState.Conv = c
diff --git a/webv2/web_test.go b/webv2/web_test.go
index 01440cc..1cff355 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/proto/migration"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
 	"github.com/cloudspannerecosystem/harbourbridge/spanner/ddl"
 	"github.com/cloudspannerecosystem/harbourbridge/webv2/session"
@@ -278,6 +279,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				ToSpanner: map[string]internal.NameAndCols{
 					"t1": {Name: "t1", Cols: map[string]string{"a": "a", "b": "b", "c": "c"}},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -487,6 +491,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				ToSpanner: map[string]internal.NameAndCols{
 					"t1": {Name: "t1", Cols: map[string]string{"a": "a", "b": "b", "c": "c"}},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -549,6 +556,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				Issues: map[string]map[string][]internal.SchemaIssue{
 					"t1": {},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -610,6 +620,9 @@ func TestUpdateTableSchema(t *testing.T) {
 				ToSource: map[string]internal.NameAndCols{
 					"t1": {Name: "t1", Cols: map[string]string{"a": "a", "b": "b", "c": "c"}},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1091,13 +1104,17 @@ func TestSetParentTable(t *testing.T) {
 				SpSchema: map[string]ddl.CreateTable{"t1": {
 					Name:     "t1",
 					ColNames: []string{"a", "b", "c"},
-					ColDefs: map[string]ddl.ColumnDef{"a": {Name: "a", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
-						"b": {Name: "b", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
-						"c": {Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true}},
-					Pks: []ddl.IndexKey{{Col: "a", Desc: false}},
-					Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"a"}, ReferTable: "ref_t1", ReferColumns: []string{"ref_c1"}},
-						{Name: "fk2", Columns: []string{"c"}, ReferTable: "ref_t2", ReferColumns: []string{"ref_c2"}}},
-				}}},
+					ColDefs: map[string]ddl.ColumnDef{"a": ddl.ColumnDef{Name: "a", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+						"b": ddl.ColumnDef{Name: "b", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+						"c": ddl.ColumnDef{Name: "c", T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}, NotNull: true}},
+					Pks: []ddl.IndexKey{ddl.IndexKey{Col: "a", Desc: false}},
+					Fks: []ddl.Foreignkey{ddl.Foreignkey{Name: "fk1", Columns: []string{"a"}, ReferTable: "ref_t1", ReferColumns: []string{"ref_c1"}},
+						ddl.Foreignkey{Name: "fk2", Columns: []string{"c"}, ReferTable: "ref_t2", ReferColumns: []string{"ref_c2"}}},
+				}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
+			},
 		},
 		{
 			name: "table with synthetic PK",
@@ -1114,7 +1131,10 @@ func TestSetParentTable(t *testing.T) {
 					Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"a"}, ReferTable: "ref_t1", ReferColumns: []string{"ref_c1"}},
 						{Name: "fk2", Columns: []string{"c"}, ReferTable: "ref_t2", ReferColumns: []string{"ref_c2"}}},
 				}},
-				SyntheticPKeys: map[string]internal.SyntheticPKey{"t1": {Col: "synth_id"}},
+				SyntheticPKeys: map[string]internal.SyntheticPKey{"t1": internal.SyntheticPKey{Col: "synth_id"}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1145,7 +1165,10 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "synth_id", Desc: false}},
 					},
 				},
-				SyntheticPKeys: map[string]internal.SyntheticPKey{"t2": {Col: "synth_id"}},
+				SyntheticPKeys: map[string]internal.SyntheticPKey{"t2": internal.SyntheticPKey{Col: "synth_id"}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1176,6 +1199,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "b", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1206,6 +1232,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1236,6 +1265,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1267,6 +1299,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "a", Desc: false}, {Col: "b", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1309,6 +1344,9 @@ func TestSetParentTable(t *testing.T) {
 						Pks: []ddl.IndexKey{{Col: "c", Desc: false}},
 					},
 				},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			table:            "t1",
 			statusCode:       http.StatusOK,
@@ -1382,6 +1420,9 @@ func TestDropForeignKey(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1400,6 +1441,9 @@ func TestDropForeignKey(t *testing.T) {
 					"t1": {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 		{
@@ -1412,6 +1456,9 @@ func TestDropForeignKey(t *testing.T) {
 					"t1": {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 	}
@@ -1462,6 +1509,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1484,6 +1534,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx_1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx_2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1506,6 +1559,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx_1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx_2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1528,6 +1584,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx_1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx_2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1552,6 +1611,9 @@ func TestRenameIndexes(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1573,6 +1635,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1595,6 +1660,9 @@ func TestRenameIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1614,6 +1682,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1632,6 +1703,9 @@ func TestRenameIndexes(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1696,6 +1770,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1719,6 +1796,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1741,6 +1821,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1763,6 +1846,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1787,6 +1873,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1810,6 +1899,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1832,6 +1924,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1852,6 +1947,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1872,6 +1970,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1896,6 +1997,9 @@ func TestRenameForeignKeys(t *testing.T) {
 						Fks: []ddl.Foreignkey{{Name: "t2_fk1", Columns: []string{"b"}, ReferTable: "reft1", ReferColumns: []string{"ref_b"}},
 							{Name: "t2_fk2", Columns: []string{"c", "d"}, ReferTable: "reft2", ReferColumns: []string{"ref_c", "ref_d"}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1962,6 +2066,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -1985,6 +2092,9 @@ func TestAddIndexes(t *testing.T) {
 							{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2011,6 +2121,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2036,6 +2149,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2060,6 +2176,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2085,6 +2204,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{ddl.IndexKey{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2107,6 +2229,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2128,6 +2253,9 @@ func TestAddIndexes(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2192,6 +2320,9 @@ func TestDropSecondaryIndex(t *testing.T) {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}},
 							{Name: "idx2", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "c", Desc: false}, {Col: "d", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 			expectedConv: &internal.Conv{
 				SpSchema: map[string]ddl.CreateTable{
@@ -2210,6 +2341,9 @@ func TestDropSecondaryIndex(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 		{
@@ -2222,6 +2356,9 @@ func TestDropSecondaryIndex(t *testing.T) {
 					"t1": {
 						Indexes: []ddl.CreateIndex{{Name: "idx1", Table: "t1", Unique: false, Keys: []ddl.IndexKey{{Col: "b", Desc: false}}}},
 					}},
+				Audit: internal.Audit{
+					MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
+				},
 			},
 		},
 	}
@@ -2355,6 +2492,7 @@ func buildConvMySQL(conv *internal.Conv) {
 		},
 	}
 	conv.SyntheticPKeys["t2"] = internal.SyntheticPKey{"synth_id", 0}
+	conv.Audit.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
 }
 
 func buildConvPostgres(conv *internal.Conv) {
@@ -2463,4 +2601,5 @@ func buildConvPostgres(conv *internal.Conv) {
 		},
 	}
 	conv.SyntheticPKeys["t2"] = internal.SyntheticPKey{"synth_id", 0}
+	conv.Audit.MigrationType = migration.MigrationData_SCHEMA_AND_DATA.Enum()
 }