Repository: cloudspannerecosystem/harbourbridge
Pull request #364 (2022-08-10 21:53:17)

Milestone5 feature drop restore table
-  Implemented drop table
-  Implemented restore table


Patch:
diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml
index d018a4c..87b69eb 100644
--- a/.github/workflows/lint.yaml
+++ b/.github/workflows/lint.yaml
@@ -28,6 +28,6 @@ jobs:
           go-version: '^1.13'
       - run: go version
       - run: go vet .
-      - run: go install golang.org/x/tools/cmd/goimports && test -z $(goimports -l .)
-      - run: go install golang.org/x/lint/golint && golint .
+      - run: go get golang.org/x/tools/cmd/goimports && test -z $(goimports -l .)
+      - run: go install golang.org/x/lint/golint@latest && golint .
       - run: go install honnef.co/go/tools/cmd/staticcheck@latest && staticcheck .
diff --git a/.gitignore b/.gitignore
index 44bdf4f..e329760 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,6 +4,7 @@ web/harbour_bridge_output/
 harbour_bridge_output/
 *__debug_bin*
 *.code-workspace
+*.log
 
 # angular ui development
 ui/.angular
@@ -11,4 +12,4 @@ ui/dist
 ui/node_modules
 
 # webv2 config
-webv2/config.json
+webv2/config.json
\ No newline at end of file
diff --git a/common/constants/constants.go b/common/constants/constants.go
index 639832c..743f5a1 100644
--- a/common/constants/constants.go
+++ b/common/constants/constants.go
@@ -61,4 +61,7 @@ const (
 
 	// Information passed in metadata while using Cloud Spanner client.
 	MigrationMetadataKey string = "cloud-spanner-migration-metadata"
+
+	// Scheme used for GCS paths
+	GCS_SCHEME string = "gs"
 )
diff --git a/common/utils/utils.go b/common/utils/utils.go
index c69d9df..a3dc670 100644
--- a/common/utils/utils.go
+++ b/common/utils/utils.go
@@ -73,7 +73,7 @@ func NewIOStreams(driver string, dumpFile string) IOStreams {
 		fmt.Printf("\nLoading dump file from path: %s\n", dumpFile)
 		var f *os.File
 		var err error
-		if u.Scheme == "gs" {
+		if u.Scheme == constants.GCS_SCHEME {
 			bucketName := u.Host
 			filePath := u.Path[1:] // removes "/" from beginning of path
 			f, err = DownloadFromGCS(bucketName, filePath, "harbourbridge.gcs.data")
@@ -152,7 +152,7 @@ func PreloadGCSFiles(tables []ManifestTable) ([]ManifestTable, error) {
 			if err != nil {
 				return nil, fmt.Errorf("unable parse file path %s for table %s", filePath, table.Table_name)
 			}
-			if u.Scheme == "gs" {
+			if u.Scheme == constants.GCS_SCHEME {
 				bucketName := u.Host
 				filePath := u.Path[1:] // removes "/" from beginning of path
 				tmpFile := strings.ReplaceAll(filePath, "/", ".")
@@ -170,6 +170,52 @@ func PreloadGCSFiles(tables []ManifestTable) ([]ManifestTable, error) {
 	return tables, nil
 }
 
+func ParseGCSFilePath(filePath string) (*url.URL, error) {
+	if len(filePath) == 0 {
+		return nil, fmt.Errorf("found empty GCS path")
+	}
+	if filePath[len(filePath)-1] != '/' {
+		filePath = filePath + "/"
+	}
+	u, err := url.Parse(filePath)
+	if err != nil {
+		return nil, fmt.Errorf("parseFilePath: unable to parse file path %s", filePath)
+	}
+	if u.Scheme != constants.GCS_SCHEME {
+		return nil, fmt.Errorf("not a valid GCS path: %s, should start with 'gs'", filePath)
+	}
+	return u, nil
+}
+
+func WriteToGCS(filePath, fileName, data string) error {
+	ctx := context.Background()
+
+	client, err := storage.NewClient(ctx)
+	if err != nil {
+		fmt.Printf("Failed to create GCS client")
+		return err
+	}
+	defer client.Close()
+	u, err := ParseGCSFilePath(filePath)
+	if err != nil {
+		return fmt.Errorf("parseFilePath: unable to parse file path: %v", err)
+	}
+	bucketName := u.Host
+	bucket := client.Bucket(bucketName)
+	obj := bucket.Object(u.Path[1:] + fileName)
+
+	w := obj.NewWriter(ctx)
+	if _, err := fmt.Fprint(w, data); err != nil {
+		fmt.Printf("Failed to write to Cloud Storage: %s", filePath)
+		return err
+	}
+	if err := w.Close(); err != nil {
+		fmt.Printf("Failed to close GCS file: %s", filePath)
+		return err
+	}
+	return nil
+}
+
 // GetProject returns the cloud project we should use for accessing Spanner.
 // Use environment variable GCLOUD_PROJECT if it is set.
 // Otherwise, use the default project returned from gcloud.
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 2156c98..01b3662 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -174,7 +174,7 @@ func schemaFromDatabase(sourceProfile profiles.SourceProfile, targetProfile prof
 	return conv, common.ProcessSchema(conv, infoSchema)
 }
 
-func performSnapshotMigration(config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client, infoSchema common.InfoSchema) (*writer.BatchWriter, error) {
+func performSnapshotMigration(config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client, infoSchema common.InfoSchema) *writer.BatchWriter {
 	common.SetRowStats(conv, infoSchema)
 	totalRows := conv.Rows()
 	var p *internal.Progress
@@ -184,7 +184,19 @@ func performSnapshotMigration(config writer.BatchWriterConfig, conv *internal.Co
 	batchWriter := populateDataConv(conv, config, client, p)
 	common.ProcessData(conv, infoSchema)
 	batchWriter.Flush()
-	return batchWriter, nil
+	return batchWriter
+}
+
+func snapshotMigrationHandler(sourceProfile profiles.SourceProfile, config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client, infoSchema common.InfoSchema) (*writer.BatchWriter, error) {
+	switch sourceProfile.Driver {
+	// Skip snapshot migration via harbourbridge for mysql and oracle since dataflow job will job will handle this from backfilled data.
+	case constants.MYSQL, constants.ORACLE:
+		return &writer.BatchWriter{}, nil
+	case constants.DYNAMODB:
+		return performSnapshotMigration(config, conv, client, infoSchema), nil
+	default:
+		return &writer.BatchWriter{}, fmt.Errorf("streaming migration not supported for driver %s", sourceProfile.Driver)
+	}
 }
 
 func dataFromDatabase(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, config writer.BatchWriterConfig, conv *internal.Conv, client *sp.Client) (*writer.BatchWriter, error) {
@@ -198,18 +210,17 @@ func dataFromDatabase(ctx context.Context, sourceProfile profiles.SourceProfile,
 		if err != nil {
 			return nil, err
 		}
-	}
-	bw, err := performSnapshotMigration(config, conv, client, infoSchema)
-	if err != nil {
-		return nil, err
-	}
-	if sourceProfile.Conn.Streaming {
+		bw, err := snapshotMigrationHandler(sourceProfile, config, conv, client, infoSchema)
+		if err != nil {
+			return nil, err
+		}
 		err = infoSchema.StartStreamingMigration(ctx, client, conv, streamInfo)
 		if err != nil {
 			return nil, err
 		}
+		return bw, nil
 	}
-	return bw, nil
+	return performSnapshotMigration(config, conv, client, infoSchema), nil
 }
 
 func getDynamoDBClientConfig() (*aws.Config, error) {
diff --git a/internal/convert.go b/internal/convert.go
index d864761..a00092b 100644
--- a/internal/convert.go
+++ b/internal/convert.go
@@ -146,8 +146,8 @@ type statementStat struct {
 // Stores the audit information of conversion.
 // 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.
+	ToSpannerFkIdx           map[string]FkeyAndIdxs                 // Maps from source-DB table name to Spanner names for table name, foreign key and indexes.
+	ToSourceFkIdx            map[string]FkeyAndIdxs                 // 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
diff --git a/internal/mapping.go b/internal/mapping.go
index 48ff178..7035b33 100644
--- a/internal/mapping.go
+++ b/internal/mapping.go
@@ -219,14 +219,17 @@ func resolveFks(conv *Conv, table string, fks []ddl.Foreignkey) []ddl.Foreignkey
 		var err error
 		if fk.Columns, err = resolveColRefs(conv, table, fk.Columns); err != nil {
 			conv.Unexpected(fmt.Sprintf("Can't resolve Columns in foreign key constraint: %s", err))
+			delete(conv.UsedNames, fk.Name)
 			continue
 		}
 		if fk.ReferTable, err = resolveTableRef(conv, fk.ReferTable); err != nil {
 			conv.Unexpected(fmt.Sprintf("Can't resolve ReferTable in foreign key constraint: %s", err))
+			delete(conv.UsedNames, fk.Name)
 			continue
 		}
 		if fk.ReferColumns, err = resolveColRefs(conv, fk.ReferTable, fk.ReferColumns); err != nil {
 			conv.Unexpected(fmt.Sprintf("Can't resolve ReferColumns in foreign key constraint: %s", err))
+			delete(conv.UsedNames, fk.Name)
 			continue
 		}
 		resolved = append(resolved, fk)
diff --git a/internal/report.go b/internal/report.go
index a33a045..23311cf 100644
--- a/internal/report.go
+++ b/internal/report.go
@@ -246,7 +246,9 @@ func AnalyzeTables(conv *Conv, badWrites map[string]int64) (r []tableReport) {
 	}
 	sort.Strings(tables)
 	for _, srcTable := range tables {
-		r = append(r, buildTableReport(conv, srcTable, badWrites))
+		if _, isPresent := conv.ToSpanner[srcTable]; isPresent {
+			r = append(r, buildTableReport(conv, srcTable, badWrites))
+		}
 	}
 	return r
 }
diff --git a/sources/common/toddl.go b/sources/common/toddl.go
index f6850f1..2dffae0 100644
--- a/sources/common/toddl.go
+++ b/sources/common/toddl.go
@@ -52,58 +52,63 @@ type ToDdl interface {
 // the Spanner schema to conv.SpSchema.
 func SchemaToSpannerDDL(conv *internal.Conv, toddl ToDdl) error {
 	for _, srcTable := range conv.SrcSchema {
-		spTableName, err := internal.GetSpannerTable(conv, srcTable.Name)
+		SchemaToSpannerDDLHelper(conv, toddl, srcTable, false)
+	}
+	internal.ResolveRefs(conv)
+	return nil
+}
+
+func SchemaToSpannerDDLHelper(conv *internal.Conv, toddl ToDdl, srcTable schema.Table, isRestore bool) error {
+	spTableName, err := internal.GetSpannerTable(conv, srcTable.Name)
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Couldn't map source table %s to Spanner: %s", srcTable.Name, err))
+		return err
+	}
+	var spColNames []string
+	spColDef := make(map[string]ddl.ColumnDef)
+	conv.Issues[srcTable.Name] = make(map[string][]internal.SchemaIssue)
+	// Iterate over columns using ColNames order.
+	for _, srcColName := range srcTable.ColNames {
+		srcCol := srcTable.ColDefs[srcColName]
+		colName, err := internal.GetSpannerCol(conv, srcTable.Name, srcCol.Name, false)
 		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Couldn't map source table %s to Spanner: %s", srcTable.Name, err))
+			conv.Unexpected(fmt.Sprintf("Couldn't map source column %s of table %s to Spanner: %s", srcTable.Name, srcCol.Name, err))
 			continue
 		}
-		var spColNames []string
-		spColDef := make(map[string]ddl.ColumnDef)
-		conv.Issues[srcTable.Name] = make(map[string][]internal.SchemaIssue)
-		// Iterate over columns using ColNames order.
-		for _, srcColName := range srcTable.ColNames {
-			srcCol := srcTable.ColDefs[srcColName]
-			colName, err := internal.GetSpannerCol(conv, srcTable.Name, srcCol.Name, false)
-			if err != nil {
-				conv.Unexpected(fmt.Sprintf("Couldn't map source column %s of table %s to Spanner: %s", srcTable.Name, srcCol.Name, err))
-				continue
-			}
-			spColNames = append(spColNames, colName)
-			ty, issues := toddl.ToSpannerType(conv, srcCol.Type)
-			// TODO(hengfeng): add issues for all elements of srcCol.Ignored.
-			if srcCol.Ignored.ForeignKey {
-				issues = append(issues, internal.ForeignKey)
-			}
-			if srcCol.Name != colName {
-				issues = append(issues, internal.IllegalName)
-			}
-			if srcCol.Ignored.Default {
-				issues = append(issues, internal.DefaultValue)
-			}
-			if srcCol.Ignored.AutoIncrement { //TODO(adibh) - check why this is not there in postgres
-				issues = append(issues, internal.AutoIncrement)
-			}
-			if len(issues) > 0 {
-				conv.Issues[srcTable.Name][srcCol.Name] = issues
-			}
-			spColDef[colName] = ddl.ColumnDef{
-				Name:    colName,
-				T:       ty,
-				NotNull: srcCol.NotNull,
-				Comment: "From: " + quoteIfNeeded(srcCol.Name) + " " + srcCol.Type.Print(),
-			}
+		spColNames = append(spColNames, colName)
+		ty, issues := toddl.ToSpannerType(conv, srcCol.Type)
+		// TODO(hengfeng): add issues for all elements of srcCol.Ignored.
+		if srcCol.Ignored.ForeignKey {
+			issues = append(issues, internal.ForeignKey)
+		}
+		if srcCol.Name != colName {
+			issues = append(issues, internal.IllegalName)
+		}
+		if srcCol.Ignored.Default {
+			issues = append(issues, internal.DefaultValue)
+		}
+		if srcCol.Ignored.AutoIncrement { //TODO(adibh) - check why this is not there in postgres
+			issues = append(issues, internal.AutoIncrement)
+		}
+		if len(issues) > 0 {
+			conv.Issues[srcTable.Name][srcCol.Name] = issues
+		}
+		spColDef[colName] = ddl.ColumnDef{
+			Name:    colName,
+			T:       ty,
+			NotNull: srcCol.NotNull,
+			Comment: "From: " + quoteIfNeeded(srcCol.Name) + " " + srcCol.Type.Print(),
 		}
-		comment := "Spanner schema for source table " + quoteIfNeeded(srcTable.Name)
-		conv.SpSchema[spTableName] = ddl.CreateTable{
-			Name:     spTableName,
-			ColNames: spColNames,
-			ColDefs:  spColDef,
-			Pks:      cvtPrimaryKeys(conv, srcTable.Name, srcTable.PrimaryKeys),
-			Fks:      cvtForeignKeys(conv, spTableName, srcTable.Name, srcTable.ForeignKeys),
-			Indexes:  cvtIndexes(conv, spTableName, srcTable.Name, srcTable.Indexes),
-			Comment:  comment}
 	}
-	internal.ResolveRefs(conv)
+	comment := "Spanner schema for source table " + quoteIfNeeded(srcTable.Name)
+	conv.SpSchema[spTableName] = ddl.CreateTable{
+		Name:     spTableName,
+		ColNames: spColNames,
+		ColDefs:  spColDef,
+		Pks:      cvtPrimaryKeys(conv, srcTable.Name, srcTable.PrimaryKeys),
+		Fks:      cvtForeignKeys(conv, spTableName, srcTable.Name, srcTable.ForeignKeys, isRestore),
+		Indexes:  cvtIndexes(conv, spTableName, srcTable.Name, srcTable.Indexes),
+		Comment:  comment}
 	return nil
 }
 
@@ -130,44 +135,57 @@ func cvtPrimaryKeys(conv *internal.Conv, srcTable string, srcKeys []schema.Key)
 	return spKeys
 }
 
-func cvtForeignKeys(conv *internal.Conv, spTableName string, srcTable string, srcKeys []schema.ForeignKey) []ddl.Foreignkey {
+func cvtForeignKeys(conv *internal.Conv, spTableName string, srcTable string, srcKeys []schema.ForeignKey, isRestore bool) []ddl.Foreignkey {
 	var spKeys []ddl.Foreignkey
 	for _, key := range srcKeys {
-		if len(key.Columns) != len(key.ReferColumns) {
-			conv.Unexpected(fmt.Sprintf("ConvertForeignKeys: columns and referColumns don't have the same lengths: len(columns)=%d, len(referColumns)=%d for source table: %s, referenced table: %s", len(key.Columns), len(key.ReferColumns), srcTable, key.ReferTable))
-			continue
-		}
-		spReferTable, err := internal.GetSpannerTable(conv, key.ReferTable)
+		spKey, err := cvtForeignKeysHelper(conv, spTableName, srcTable, key, isRestore)
 		if err != nil {
-			conv.Unexpected(fmt.Sprintf("Can't map foreign key for source table: %s, referenced table: %s", srcTable, key.ReferTable))
 			continue
 		}
-		var spCols, spReferCols []string
-		for i, col := range key.Columns {
-			spCol, err1 := internal.GetSpannerCol(conv, srcTable, col, false)
-			spReferCol, err2 := internal.GetSpannerCol(conv, key.ReferTable, key.ReferColumns[i], false)
-			if err1 != nil || err2 != nil {
-				conv.Unexpected(fmt.Sprintf("Can't map foreign key for table: %s, referenced table: %s, column: %s", srcTable, key.ReferTable, col))
-				continue
-			}
-			spCols = append(spCols, spCol)
-			spReferCols = append(spReferCols, spReferCol)
-		}
-		spKeyName := internal.ToSpannerForeignKey(conv, key.Name)
-
-		spKey := ddl.Foreignkey{
-			Name:         spKeyName,
-			Columns:      spCols,
-			ReferTable:   spReferTable,
-			ReferColumns: spReferCols}
 		spKeys = append(spKeys, spKey)
-		conv.Audit.ToSpannerFkIdx[srcTable].ForeignKey[key.Name] = spKeyName
-		conv.Audit.ToSourceFkIdx[spTableName].ForeignKey[spKeyName] = key.Name
 	}
-
 	return spKeys
 }
 
+func cvtForeignKeysHelper(conv *internal.Conv, spTableName string, srcTable string, srcKey schema.ForeignKey, isRestore bool) (ddl.Foreignkey, error) {
+	if len(srcKey.Columns) != len(srcKey.ReferColumns) {
+		conv.Unexpected(fmt.Sprintf("ConvertForeignKeys: columns and referColumns don't have the same lengths: len(columns)=%d, len(referColumns)=%d for source table: %s, referenced table: %s", len(srcKey.Columns), len(srcKey.ReferColumns), srcTable, srcKey.ReferTable))
+		return ddl.Foreignkey{}, fmt.Errorf("ConvertForeignKeys: columns and referColumns don't have the same lengths")
+	}
+	_, isPresent := conv.UsedNames[srcKey.ReferTable]
+	if !isPresent && isRestore {
+		return ddl.Foreignkey{}, nil
+	}
+	spReferTable, err := internal.GetSpannerTable(conv, srcKey.ReferTable)
+
+	if err != nil {
+		conv.Unexpected(fmt.Sprintf("Can't map foreign key for source table: %s, referenced table: %s", srcTable, srcKey.ReferTable))
+		return ddl.Foreignkey{}, err
+	}
+	var spCols, spReferCols []string
+	for i, col := range srcKey.Columns {
+		spCol, err1 := internal.GetSpannerCol(conv, srcTable, col, false)
+		spReferCol, err2 := internal.GetSpannerCol(conv, srcKey.ReferTable, srcKey.ReferColumns[i], false)
+		if err1 != nil || err2 != nil {
+			conv.Unexpected(fmt.Sprintf("Can't map foreign key for table: %s, referenced table: %s, column: %s", srcTable, srcKey.ReferTable, col))
+			continue
+		}
+		spCols = append(spCols, spCol)
+		spReferCols = append(spReferCols, spReferCol)
+	}
+	spKeyName := internal.ToSpannerForeignKey(conv, srcKey.Name)
+
+	spKey := ddl.Foreignkey{
+		Name:         spKeyName,
+		Columns:      spCols,
+		ReferTable:   spReferTable,
+		ReferColumns: spReferCols}
+	conv.Audit.ToSpannerFkIdx[srcTable].ForeignKey[srcKey.Name] = spKeyName
+	conv.Audit.ToSourceFkIdx[spTableName].ForeignKey[spKeyName] = srcKey.Name
+
+	return spKey, nil
+}
+
 func cvtIndexes(conv *internal.Conv, spTableName string, srcTable string, srcIndexes []schema.Index) []ddl.CreateIndex {
 	var spIndexes []ddl.CreateIndex
 	for _, srcIndex := range srcIndexes {
@@ -209,3 +227,39 @@ func cvtIndexes(conv *internal.Conv, spTableName string, srcTable string, srcInd
 	}
 	return spIndexes
 }
+
+func SrcTableToSpannerDDL(conv *internal.Conv, toddl ToDdl, srcTable schema.Table) error {
+	err := SchemaToSpannerDDLHelper(conv, toddl, srcTable, true)
+	if err != nil {
+		return err
+	}
+	for srcTableName, sourceTable := range conv.SrcSchema {
+		if _, isPresent := conv.ToSpanner[srcTableName]; !isPresent {
+			continue
+		}
+		spTableName := conv.ToSpanner[srcTableName].Name
+		if _, isPresent := conv.SpSchema[spTableName]; !isPresent {
+			continue
+		}
+		spTable := conv.SpSchema[spTableName]
+		if sourceTable.Name != srcTable.Name {
+			spTable.Fks = cvtForeignKeysForAReferenceTable(conv, spTableName, srcTableName, srcTable.Name, sourceTable.ForeignKeys, spTable.Fks)
+			conv.SpSchema[spTableName] = spTable
+		}
+	}
+	internal.ResolveRefs(conv)
+	return nil
+}
+
+func cvtForeignKeysForAReferenceTable(conv *internal.Conv, spTableName string, srcTable string, referTable string, srcKeys []schema.ForeignKey, spKeys []ddl.Foreignkey) []ddl.Foreignkey {
+	for _, key := range srcKeys {
+		if key.ReferTable == referTable {
+			spKey, err := cvtForeignKeysHelper(conv, spTableName, srcTable, key, true)
+			if err != nil {
+				continue
+			}
+			spKeys = append(spKeys, spKey)
+		}
+	}
+	return spKeys
+}
diff --git a/sources/mysql/infoschema.go b/sources/mysql/infoschema.go
index 2542877..97c5043 100644
--- a/sources/mysql/infoschema.go
+++ b/sources/mysql/infoschema.go
@@ -17,6 +17,7 @@ package mysql
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"sort"
 	"strings"
@@ -25,6 +26,7 @@ import (
 	_ "github.com/go-sql-driver/mysql" // The driver should be used via the database/sql package.
 	_ "github.com/lib/pq"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/schema"
@@ -350,8 +352,22 @@ func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *inte
 // performing a streaming migration.
 func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, streamingInfo map[string]interface{}) error {
 	streamingCfg, _ := streamingInfo["streamingCfg"].(streaming.StreamingCfg)
-	err := streaming.StartDataflow(ctx, isi.SourceProfile, isi.TargetProfile, streamingCfg)
+	convJSON, err := json.MarshalIndent(conv, "", " ")
 	if err != nil {
+		err = fmt.Errorf("can't encode session state to JSON: %v", err)
+		return err
+	}
+	fmt.Printf("Writing session file to GCS...")
+	err = utils.WriteToGCS(streamingCfg.TmpDir, "session.json", string(convJSON))
+	if err != nil {
+		err = fmt.Errorf("error writing session file to GCS: %v", err)
+		return err
+	}
+	fmt.Println("Done")
+
+	err = streaming.StartDataflow(ctx, isi.SourceProfile, isi.TargetProfile, streamingCfg)
+	if err != nil {
+		err = fmt.Errorf("error starting dataflow: %v", err)
 		return err
 	}
 	return nil
diff --git a/streaming/streaming.go b/streaming/streaming.go
index ae8d7dc..eed26f4 100644
--- a/streaming/streaming.go
+++ b/streaming/streaming.go
@@ -18,10 +18,12 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"strings"
 	"time"
 
 	dataflow "cloud.google.com/go/dataflow/apiv1beta3"
 	datastream "cloud.google.com/go/datastream/apiv1alpha1"
+	"cloud.google.com/go/storage"
 	datastreampb "google.golang.org/genproto/googleapis/cloud/datastream/v1alpha1"
 	dataflowpb "google.golang.org/genproto/googleapis/dataflow/v1beta3"
 	fieldmaskpb "google.golang.org/protobuf/types/known/fieldmaskpb"
@@ -31,6 +33,11 @@ import (
 	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 )
 
+const (
+	numWorkers int32 = 50
+	maxWorkers int32 = 50
+)
+
 type SrcConnCfg struct {
 	Name     string
 	Location string
@@ -58,6 +65,7 @@ type DataflowCfg struct {
 type StreamingCfg struct {
 	DatastreamCfg DatastreamCfg
 	DataflowCfg   DataflowCfg
+	TmpDir        string
 }
 
 // VerifyAndUpdateCfg checks the fields and errors out if certain fields are empty.
@@ -100,11 +108,30 @@ func VerifyAndUpdateCfg(streamingCfg *StreamingCfg, dbName string) error {
 	if dfCfg.JobName == "" {
 		// Update names to have more info like dbname.
 		jobName, err := utils.GenerateName("hb-dataflow-" + dbName)
+		jobName = strings.Replace(jobName, "_", "-", -1)
 		if err != nil {
 			return fmt.Errorf("error generating stream name: %v", err)
 		}
 		streamingCfg.DataflowCfg.JobName = jobName
 	}
+
+	filePath := streamingCfg.TmpDir
+	u, err := utils.ParseGCSFilePath(filePath)
+	if err != nil {
+		return fmt.Errorf("parseFilePath: unable to parse file path: %v", err)
+	}
+	bucketName := u.Host
+	ctx := context.Background()
+	client, err := storage.NewClient(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create GCS client")
+	}
+	defer client.Close()
+	bucket := client.Bucket(bucketName)
+	_, err = bucket.Attrs(ctx)
+	if err != nil {
+		return fmt.Errorf("bucket %s does not exist", bucketName)
+	}
 	return nil
 }
 
@@ -187,7 +214,7 @@ func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, pro
 		SourceConfig:      srcCfg,
 		DestinationConfig: dstCfg,
 		State:             datastreampb.Stream_RUNNING,
-		BackfillStrategy:  &datastreampb.Stream_BackfillNone{BackfillNone: &datastreampb.Stream_BackfillNoneStrategy{}},
+		BackfillStrategy:  &datastreampb.Stream_BackfillAll{BackfillAll: &datastreampb.Stream_BackfillAllStrategy{}},
 	}
 	createStreamRequest := &datastreampb.CreateStreamRequest{
 		Parent:   fmt.Sprintf("projects/%s/locations/%s", projectID, datastreamCfg.StreamLocation),
@@ -199,13 +226,13 @@ func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, pro
 
 	dsOp, err := dsClient.CreateStream(ctx, createStreamRequest)
 	if err != nil {
-		fmt.Printf("createStreamRequest: %+v\n", createStreamRequest)
+		fmt.Printf("cannot create stream: createStreamRequest: %+v\n", createStreamRequest)
 		return fmt.Errorf("cannot create stream: %v ", err)
 	}
 
 	_, err = dsOp.Wait(ctx)
 	if err != nil {
-		fmt.Printf("createStreamRequest: %+v\n", createStreamRequest)
+		fmt.Printf("datastream create operation failed: createStreamRequest: %+v\n", createStreamRequest)
 		return fmt.Errorf("datastream create operation failed: %v", err)
 	}
 	fmt.Println("Successfully created stream ", datastreamCfg.StreamId)
@@ -229,8 +256,10 @@ func LaunchStream(ctx context.Context, sourceProfile profiles.SourceProfile, pro
 }
 
 // LaunchDataflowJob populates the parameters from the streaming config and triggers a Dataflow job.
-func LaunchDataflowJob(ctx context.Context, targetProfile profiles.TargetProfile, datastreamCfg DatastreamCfg, dataflowCfg DataflowCfg) error {
+func LaunchDataflowJob(ctx context.Context, targetProfile profiles.TargetProfile, streamingCfg StreamingCfg) error {
 	project, instance, dbName, _ := targetProfile.GetResourceIds(ctx, time.Now(), "", nil)
+	dataflowCfg := streamingCfg.DataflowCfg
+	datastreamCfg := streamingCfg.DatastreamCfg
 	fmt.Println("Launching dataflow job ", dataflowCfg.JobName, " in ", project, "-", dataflowCfg.Location)
 
 	c, err := dataflow.NewFlexTemplatesClient(ctx)
@@ -265,6 +294,11 @@ func LaunchDataflowJob(ctx context.Context, targetProfile profiles.TargetProfile
 			"streamName":       fmt.Sprintf("projects/%s/locations/%s/streams/%s", project, datastreamCfg.StreamLocation, datastreamCfg.StreamId),
 			"instanceId":       instance,
 			"databaseId":       dbName,
+			"sessionFilePath":  streamingCfg.TmpDir + "session.json",
+		},
+		Environment: &dataflowpb.FlexTemplateRuntimeEnvironment{
+			NumWorkers: numWorkers,
+			MaxWorkers: maxWorkers,
 		},
 	}
 
@@ -313,7 +347,7 @@ func StartDatastream(ctx context.Context, sourceProfile profiles.SourceProfile,
 }
 
 func StartDataflow(ctx context.Context, sourceProfile profiles.SourceProfile, targetProfile profiles.TargetProfile, streamingCfg StreamingCfg) error {
-	err := LaunchDataflowJob(ctx, targetProfile, streamingCfg.DatastreamCfg, streamingCfg.DataflowCfg)
+	err := LaunchDataflowJob(ctx, targetProfile, streamingCfg)
 	if err != nil {
 		return fmt.Errorf("error launching dataflow: %v", err)
 	}
diff --git a/ui/src/app/app.constants.ts b/ui/src/app/app.constants.ts
index fb107e6..215b82a 100644
--- a/ui/src/app/app.constants.ts
+++ b/ui/src/app/app.constants.ts
@@ -34,5 +34,10 @@ export enum RulesTypes {
 export enum MigrationModes {
   schemaOnly = 'Schema',
   dataOnly = 'Data',
-  schemaAndData = 'Schema And Data'
+  schemaAndData = 'Schema And Data',
+}
+
+export enum ObjectDetailNodeType {
+  Table = 'table',
+  Index = 'index',
 }
diff --git a/ui/src/app/app.module.ts b/ui/src/app/app.module.ts
index b3702d6..60d6846 100644
--- a/ui/src/app/app.module.ts
+++ b/ui/src/app/app.module.ts
@@ -30,9 +30,9 @@ import { AddIndexFormComponent } from './components/add-index-form/add-index-for
 import { EditGlobalDatatypeFormComponent } from './components/edit-global-datatype-form/edit-global-datatype-form.component'
 import { SidenavViewAssessmentComponent } from './components/sidenav-view-assessment/sidenav-view-assessment.component'
 import { SidenavSaveSessionComponent } from './components/sidenav-save-session/sidenav-save-session.component'
-import { DropIndexDialogComponent } from './components/drop-index-dialog/drop-index-dialog.component'
-import { DatabaseLoaderComponent } from './components/database-loader/database-loader.component';
-import { PrepareMigrationComponent } from './components/prepare-migration/prepare-migration.component';
+import { DropIndexOrTableDialogComponent } from './components/drop-index-or-table-dialog/drop-index-or-table-dialog.component'
+import { DatabaseLoaderComponent } from './components/database-loader/database-loader.component'
+import { PrepareMigrationComponent } from './components/prepare-migration/prepare-migration.component'
 import { TargetDetailsFormComponent } from './components/target-details-form/target-details-form.component'
 
 @NgModule({
@@ -60,8 +60,8 @@ import { TargetDetailsFormComponent } from './components/target-details-form/tar
     EditGlobalDatatypeFormComponent,
     SidenavViewAssessmentComponent,
     SidenavSaveSessionComponent,
-    DropIndexDialogComponent,
     DatabaseLoaderComponent,
+    DropIndexOrTableDialogComponent,
     PrepareMigrationComponent,
     TargetDetailsFormComponent,
   ],
diff --git a/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.spec.ts b/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.spec.ts
deleted file mode 100644
index 050e215..0000000
--- a/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.spec.ts
+++ /dev/null
@@ -1,25 +0,0 @@
-import { ComponentFixture, TestBed } from '@angular/core/testing';
-
-import { DropIndexDialogComponent } from './drop-index-dialog.component';
-
-describe('DropIndexDialogComponent', () => {
-  let component: DropIndexDialogComponent;
-  let fixture: ComponentFixture<DropIndexDialogComponent>;
-
-  beforeEach(async () => {
-    await TestBed.configureTestingModule({
-      declarations: [ DropIndexDialogComponent ]
-    })
-    .compileComponents();
-  });
-
-  beforeEach(() => {
-    fixture = TestBed.createComponent(DropIndexDialogComponent);
-    component = fixture.componentInstance;
-    fixture.detectChanges();
-  });
-
-  it('should create', () => {
-    expect(component).toBeTruthy();
-  });
-});
diff --git a/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.ts b/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.ts
deleted file mode 100644
index 77ffaba..0000000
--- a/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.ts
+++ /dev/null
@@ -1,28 +0,0 @@
-import { Component, Inject, OnInit } from '@angular/core'
-import { FormControl, Validators } from '@angular/forms'
-import { MatDialogRef, MAT_DIALOG_DATA } from '@angular/material/dialog'
-
-@Component({
-  selector: 'app-drop-index-dialog',
-  templateUrl: './drop-index-dialog.component.html',
-  styleUrls: ['./drop-index-dialog.component.scss'],
-})
-export class DropIndexDialogComponent implements OnInit {
-  confirmationInput: FormControl
-  constructor(
-    @Inject(MAT_DIALOG_DATA) public data: string,
-    private dialogRef: MatDialogRef<DropIndexDialogComponent>
-  ) {
-    ;(this.confirmationInput = new FormControl('', [
-      Validators.required,
-      Validators.pattern(`^${data}$`),
-    ])),
-      (dialogRef.disableClose = true)
-  }
-
-  deleteIndex() {
-    this.dialogRef.close('delete')
-  }
-
-  ngOnInit(): void {}
-}
diff --git a/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.html b/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.html
similarity index 71%
rename from ui/src/app/components/drop-index-dialog/drop-index-dialog.component.html
rename to ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.html
index 46bb06d..e83ffce 100644
--- a/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.html
+++ b/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.html
@@ -1,16 +1,18 @@
 <div mat-dialog-content>
-  <h2>Delete Index ({{ data }})?</h2>
-  <div class="alert-container">
+  <h2>Delete {{ data.type }} ({{ data.name }})?</h2>
+  <div class="alert-container" *ngIf="data.type === 'Index'">
     <mat-icon>warning</mat-icon>
     <span>This operation cannot be undone</span>
   </div>
   <div class="form-container">
     <form>
       <span class="form-custom-label">
-        Confirm deletion by typing the following below: <b>{{ data }}</b>
+        Confirm deletion by typing the following below: <b>{{ data.name }}</b>
       </span>
       <mat-form-field appearance="outline">
-        <mat-label>Index Name</mat-label>
+        <mat-label
+          >{{ data.type == ObjectDetailNodeType.Table ? 'Table' : 'Index' }} Name</mat-label
+        >
         <input matInput type="text" [formControl]="confirmationInput" />
       </mat-form-field>
     </form>
@@ -20,7 +22,7 @@
   <button mat-button color="primary" mat-dialog-close>Cancel</button>
   <button
     mat-button
-    (click)="deleteIndex()"
+    (click)="delete()"
     type="submit"
     color="primary"
     [disabled]="!confirmationInput.valid"
diff --git a/ui/src/app/components/drop-index-dialog/drop-index-dialog.component.scss b/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.scss
similarity index 100%
rename from ui/src/app/components/drop-index-dialog/drop-index-dialog.component.scss
rename to ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.scss
diff --git a/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.spec.ts b/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.spec.ts
new file mode 100644
index 0000000..cb06df7
--- /dev/null
+++ b/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.spec.ts
@@ -0,0 +1,24 @@
+import { ComponentFixture, TestBed } from '@angular/core/testing'
+
+import { DropIndexOrTableDialogComponent } from './drop-index-or-table-dialog.component'
+
+describe('DropIndexOrTableDialogComponent', () => {
+  let component: DropIndexOrTableDialogComponent
+  let fixture: ComponentFixture<DropIndexOrTableDialogComponent>
+
+  beforeEach(async () => {
+    await TestBed.configureTestingModule({
+      declarations: [DropIndexOrTableDialogComponent],
+    }).compileComponents()
+  })
+
+  beforeEach(() => {
+    fixture = TestBed.createComponent(DropIndexOrTableDialogComponent)
+    component = fixture.componentInstance
+    fixture.detectChanges()
+  })
+
+  it('should create', () => {
+    expect(component).toBeTruthy()
+  })
+})
diff --git a/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.ts b/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.ts
new file mode 100644
index 0000000..1f4186b
--- /dev/null
+++ b/ui/src/app/components/drop-index-or-table-dialog/drop-index-or-table-dialog.component.ts
@@ -0,0 +1,30 @@
+import { Component, Inject, OnInit } from '@angular/core'
+import { FormControl, Validators } from '@angular/forms'
+import { MatDialogRef, MAT_DIALOG_DATA } from '@angular/material/dialog'
+import { ObjectDetailNodeType } from 'src/app/app.constants'
+
+@Component({
+  selector: 'app-drop-index-dialog',
+  templateUrl: './drop-index-or-table-dialog.component.html',
+  styleUrls: ['./drop-index-or-table-dialog.component.scss'],
+})
+export class DropIndexOrTableDialogComponent implements OnInit {
+  ObjectDetailNodeType = ObjectDetailNodeType
+  confirmationInput: FormControl
+  constructor(
+    @Inject(MAT_DIALOG_DATA) public data: { name: string; type: string },
+    private dialogRef: MatDialogRef<DropIndexOrTableDialogComponent>
+  ) {
+    ;(this.confirmationInput = new FormControl('', [
+      Validators.required,
+      Validators.pattern(`^${data.name}$`),
+    ])),
+      (dialogRef.disableClose = true)
+  }
+
+  delete() {
+    this.dialogRef.close(this.data.type)
+  }
+
+  ngOnInit(): void {}
+}
diff --git a/ui/src/app/components/object-detail/object-detail.component.html b/ui/src/app/components/object-detail/object-detail.component.html
index 450e8d6..88395c8 100644
--- a/ui/src/app/components/object-detail/object-detail.component.html
+++ b/ui/src/app/components/object-detail/object-detail.component.html
@@ -72,8 +72,32 @@
         <mat-icon>delete</mat-icon>
         <span>DROP INDEX</span>
       </button>
+      <button
+        mat-button
+        color="primary"
+        class="icon drop"
+        (click)="dropTable()"
+        *ngIf="
+          currentObject!.isSpannerNode &&
+          currentObject!.type !== 'indexName' &&
+          !currentObject!.isDeleted
+        "
+      >
+        <mat-icon>delete</mat-icon>
+        <span>DROP TABLE</span>
+      </button>
+      <button
+        mat-button
+        color="primary"
+        class="icon drop"
+        (click)="restoreSpannerTable()"
+        *ngIf="currentObject!.isSpannerNode && currentObject!.isDeleted"
+      >
+        <mat-icon>undo</mat-icon>
+        <span> RESTORE TABLE</span>
+      </button>
     </h3>
-    <div class="interleaved-title" *ngIf="interleaveParentName">
+    <div class="interleaved-title" *ngIf="interleaveParentName && currentObject.isSpannerNode">
       Interleaved:
       <div class="interleave-parent-name">
         {{ interleaveParentName }}
@@ -97,7 +121,7 @@
             <th mat-header-cell *matHeaderCellDef class="db_name" colspan="6">
               <div class="spanner_edit-button">
                 <span>Spanner</span>
-                <div *ngIf="!isEditMode">
+                <div *ngIf="!isEditMode && !currentObject.isDeleted">
                   <button
                     mat-stroked-button
                     color="primary"
@@ -304,7 +328,7 @@
             <th mat-header-cell *matHeaderCellDef class="db_name" colspan="5">
               <div class="spanner_edit-button">
                 <span>Spanner</span>
-                <div *ngIf="!isPkEditMode">
+                <div *ngIf="!isPkEditMode && !currentObject.isDeleted">
                   <button
                     mat-stroked-button
                     color="primary"
@@ -511,7 +535,15 @@
             <th mat-header-cell *matHeaderCellDef class="db_name" colspan="5">
               <div class="spanner_edit-button">
                 <span>Spanner</span>
-                <div *ngIf="!isFkEditMode && fkDataSource.length > 0 && !interleaveParentName && currentObject!.isSpannerNode ">
+                <div
+                  *ngIf="
+                    !isFkEditMode &&
+                    fkDataSource.length > 0 &&
+                    !interleaveParentName &&
+                    currentObject!.isSpannerNode &&
+                    !currentObject.isDeleted
+                  "
+                >
                   <button mat-stroked-button color="primary" (click)="toggleFkEdit()">
                     <mat-icon class="edit-icon">edit</mat-icon>
                     EDIT
@@ -653,9 +685,10 @@
     <mat-tab
       label="INTERLEAVE"
       *ngIf="
-        (interleaveStatus.tableInterleaveStatus &&
+        ((interleaveStatus.tableInterleaveStatus &&
           interleaveStatus.tableInterleaveStatus.Possible) ||
-        this.interleaveParentName !== null
+          this.interleaveParentName !== null) &&
+        currentObject.isSpannerNode
       "
     >
       <div class="interleave-tab-container">
@@ -686,7 +719,7 @@
       </div>
     </mat-tab>
 
-    <mat-tab label="SQL">
+    <mat-tab label="SQL" *ngIf="currentObject!.isSpannerNode && !currentObject!.isDeleted">
       <div class="ddl_container">
         <pre><code>{{ddlStmts[currentObject!.name]}}</code></pre>
       </div>
diff --git a/ui/src/app/components/object-detail/object-detail.component.ts b/ui/src/app/components/object-detail/object-detail.component.ts
index d166483..0ab6e4c 100644
--- a/ui/src/app/components/object-detail/object-detail.component.ts
+++ b/ui/src/app/components/object-detail/object-detail.component.ts
@@ -7,13 +7,13 @@ import { InfodialogComponent } from '../infodialog/infodialog.component'
 import IColumnTabData, { IIndexData } from '../../model/edit-table'
 import { SnackbarService } from 'src/app/services/snackbar/snackbar.service'
 import IFkTabData from 'src/app/model/fk-tab-data'
-import { ObjectExplorerNodeType, StorageKeys } from 'src/app/app.constants'
+import { ObjectDetailNodeType, ObjectExplorerNodeType, StorageKeys } from 'src/app/app.constants'
 import FlatNode from 'src/app/model/schema-object-node'
 import { Subscription, take } from 'rxjs'
 import { MatTabChangeEvent } from '@angular/material/tabs/tab-group'
 import IConv, { ICreateIndex, IPrimaryKey } from 'src/app/model/conv'
-import { DropIndexDialogComponent } from '../drop-index-dialog/drop-index-dialog.component'
 import { ConversionService } from 'src/app/services/conversion/conversion.service'
+import { DropIndexOrTableDialogComponent } from '../drop-index-or-table-dialog/drop-index-or-table-dialog.component'
 
 @Component({
   selector: 'app-object-detail',
@@ -403,7 +403,7 @@ export class ObjectDetailComponent implements OnInit {
   setPkOrder() {
     if (
       this.currentObject &&
-      this.conv.SpSchema[this.currentObject!.name].Pks.length == this.pkData.length
+      this.conv.SpSchema[this.currentObject!.name]?.Pks.length == this.pkData.length
     ) {
       this.pkData.forEach((pk: IColumnTabData, i: number) => {
         if (this.pkData[i].spColName === this.conv.SpSchema[this.currentObject!.name].Pks[i].Col) {
@@ -417,11 +417,11 @@ export class ObjectDetailComponent implements OnInit {
       })
     } else {
       this.pkData.forEach((pk: IColumnTabData, i: number) => {
-        let index = this.conv.SpSchema[this.currentObject!.name].Pks.map(
+        let index = this.conv.SpSchema[this.currentObject!.name]?.Pks.map(
           (item) => item.Col
         ).indexOf(pk.spColName)
         if (index !== -1) {
-          pk.spOrder = this.conv.SpSchema[this.currentObject!.name].Pks[index].Order
+          pk.spOrder = this.conv.SpSchema[this.currentObject!.name]?.Pks[index].Order
         }
       })
     }
@@ -609,7 +609,7 @@ export class ObjectDetailComponent implements OnInit {
   }
 
   checkIsInterleave() {
-    if (this.currentObject) {
+    if (!this.currentObject?.isDeleted && this.currentObject?.isSpannerNode) {
       this.data.getInterleaveConversionForATable(this.currentObject!.name)
     }
   }
@@ -623,7 +623,7 @@ export class ObjectDetailComponent implements OnInit {
     let ddl: string = ''
     if (
       this.currentObject?.type === ObjectExplorerNodeType.Table &&
-      this.ddlStmts[this.currentObject.name].includes(substr)
+      this.ddlStmts[this.currentObject.name]?.includes(substr)
     ) {
       ddl = this.ddlStmts[this.currentObject.name].substring(
         this.ddlStmts[this.currentObject.name].indexOf(substr) + 20
@@ -698,14 +698,14 @@ export class ObjectDetailComponent implements OnInit {
   }
 
   dropIndex() {
-    let openDialog = this.dialog.open(DropIndexDialogComponent, {
+    let openDialog = this.dialog.open(DropIndexOrTableDialogComponent, {
       width: '35vw',
       minWidth: '450px',
       maxWidth: '600px',
-      data: this.currentObject?.name,
+      data: { name: this.currentObject?.name, type: ObjectDetailNodeType.Index },
     })
     openDialog.afterClosed().subscribe((res: string) => {
-      if (res) {
+      if (res === ObjectDetailNodeType.Index) {
         this.data
           .dropIndex(this.currentObject!.parent, this.currentObject!.name)
           .pipe(take(1))
@@ -741,10 +741,48 @@ export class ObjectDetailComponent implements OnInit {
       srcDesc: undefined,
       srcOrder: '',
     })
-
     this.setIndexRows()
   }
 
+  restoreSpannerTable() {
+    let tableId = this.currentObject!.id
+    this.data
+      .restoreTable(tableId)
+      .pipe(take(1))
+      .subscribe((res: string) => {
+        if (res === '') {
+          this.isObjectSelected = false
+        }
+      })
+    this.currentObject = null
+  }
+
+  dropTable() {
+    let openDialog = this.dialog.open(DropIndexOrTableDialogComponent, {
+      width: '35vw',
+      minWidth: '450px',
+      maxWidth: '600px',
+      data: { name: this.currentObject?.name, type: ObjectDetailNodeType.Table },
+    })
+    openDialog.afterClosed().subscribe((res: string) => {
+      if (res === ObjectDetailNodeType.Table) {
+        let tableId = this.currentObject!.id
+        this.data
+          .dropTable(tableId)
+          .pipe(take(1))
+          .subscribe((res: string) => {
+            if (res === '') {
+              this.isObjectSelected = false
+              this.updateSidebar.emit(true)
+            }
+          })
+        this.currentObject = null
+      }
+    })
+  }
+
+  selectedColumnChange(tableName: string) {}
+
   tabChanged(tabChangeEvent: MatTabChangeEvent): void {
     this.currentTabIndex = tabChangeEvent.index
   }
diff --git a/ui/src/app/components/object-explorer/object-explorer.component.html b/ui/src/app/components/object-explorer/object-explorer.component.html
index 44d95df..fdb6e75 100644
--- a/ui/src/app/components/object-explorer/object-explorer.component.html
+++ b/ui/src/app/components/object-explorer/object-explorer.component.html
@@ -219,6 +219,14 @@
               >
                 error
               </mat-icon>
+              <mat-icon
+                *ngIf="data.status === 'DARK'"
+                class="icon dark"
+                matTooltip="Deleted"
+                matTooltipPosition="above"
+              >
+                delete
+              </mat-icon>
             </td>
           </ng-container>
           <ng-container matColumnDef="name">
@@ -243,7 +251,11 @@
                 >
               </div>
             </th>
-            <td mat-cell *matCellDef="let data" class="sidebar_link">
+            <td
+              mat-cell
+              *matCellDef="let data"
+              [ngClass]="{ sidebar_link: true, 'gray-out': data.isDeleted }"
+            >
               <button
                 mat-icon-button
                 [style.visibility]="!data.expandable ? 'hidden' : ''"
diff --git a/ui/src/app/components/object-explorer/object-explorer.component.scss b/ui/src/app/components/object-explorer/object-explorer.component.scss
index 28bbfd8..dcf8aa0 100644
--- a/ui/src/app/components/object-explorer/object-explorer.component.scss
+++ b/ui/src/app/components/object-explorer/object-explorer.component.scss
@@ -87,22 +87,6 @@
   .icon {
     margin-right: 20px;
   }
-
-  .danger {
-    color: #da4236;
-  }
-
-  .warning {
-    color: #f3b300;
-  }
-
-  .success {
-    color: green;
-  }
-
-  .orange {
-    color: rgb(248, 116, 68);
-  }
 }
 
 .filter-wrapper {
diff --git a/ui/src/app/components/object-explorer/object-explorer.component.ts b/ui/src/app/components/object-explorer/object-explorer.component.ts
index a7f7111..15b2f46 100644
--- a/ui/src/app/components/object-explorer/object-explorer.component.ts
+++ b/ui/src/app/components/object-explorer/object-explorer.component.ts
@@ -39,6 +39,9 @@ export class ObjectExplorerComponent implements OnInit {
       pos: node.pos,
       isSpannerNode: node.isSpannerNode,
       level: level,
+      isDeleted: node.isDeleted ? true : false,
+      id: node.id,
+      parentId: node.parentId,
     }
   }
   treeControl = new FlatTreeControl<FlatNode>(
diff --git a/ui/src/app/components/summary/summary.component.ts b/ui/src/app/components/summary/summary.component.ts
index a76a2b2..928febd 100644
--- a/ui/src/app/components/summary/summary.component.ts
+++ b/ui/src/app/components/summary/summary.component.ts
@@ -69,6 +69,9 @@ export class SummaryComponent implements OnInit {
         this.changeIssuesLabel.emit(
           s.NotesCount + s.WarningsCount + s.ErrorsCount + s.SuggestionsCount
         )
+      } else {
+        this.summaryCount = 0
+        this.changeIssuesLabel.emit(0)
       }
     } else {
       this.summaryCount = 0
diff --git a/ui/src/app/components/workspace/workspace.component.ts b/ui/src/app/components/workspace/workspace.component.ts
index 09ac86a..e906b6f 100644
--- a/ui/src/app/components/workspace/workspace.component.ts
+++ b/ui/src/app/components/workspace/workspace.component.ts
@@ -65,6 +65,15 @@ export class WorkspaceComponent implements OnInit, OnDestroy {
 
     this.convObj = this.data.conv.subscribe((data: IConv) => {
       const indexAddedOrRemoved = this.isIndexAddedOrRemoved(data)
+      if (
+        data &&
+        this.conv &&
+        Object.keys(data?.SpSchema).length != Object.keys(this.conv?.SpSchema).length
+      ) {
+        this.conv = data
+        this.reRenderObjectExplorerSpanner()
+        this.reRenderObjectExplorerSrc()
+      }
       this.conv = data
       if (indexAddedOrRemoved && this.conversionRates) this.reRenderObjectExplorerSpanner()
       if (!this.objectExplorerInitiallyRender && this.conversionRates) {
@@ -74,11 +83,11 @@ export class WorkspaceComponent implements OnInit, OnDestroy {
       }
       if (this.currentObject && this.currentObject.type === ObjectExplorerNodeType.Table) {
         this.fkData = this.currentObject
-          ? this.conversion.getFkMapping(this.currentObject.name, data)
+          ? this.conversion.getFkMapping(this.currentObject.id, data)
           : []
 
         this.tableData = this.currentObject
-          ? this.conversion.getColumnMapping(this.currentObject.name, data)
+          ? this.conversion.getColumnMapping(this.currentObject.id, data)
           : []
       }
       if (
@@ -87,9 +96,9 @@ export class WorkspaceComponent implements OnInit, OnDestroy {
         !indexAddedOrRemoved
       ) {
         this.indexData = this.conversion.getIndexMapping(
-          this.currentObject.parent,
+          this.currentObject.parentId,
           this.conv,
-          this.currentObject.name
+          this.currentObject.id
         )
       }
     })
@@ -157,16 +166,16 @@ export class WorkspaceComponent implements OnInit, OnDestroy {
     if (object?.type === ObjectExplorerNodeType.Table) {
       this.currentObject = object
       this.tableData = this.currentObject
-        ? this.conversion.getColumnMapping(this.currentObject.name, this.conv)
+        ? this.conversion.getColumnMapping(this.currentObject.id, this.conv)
         : []
 
       this.fkData = []
       this.fkData = this.currentObject
-        ? this.conversion.getFkMapping(this.currentObject.name, this.conv)
+        ? this.conversion.getFkMapping(this.currentObject.id, this.conv)
         : []
     } else if (object?.type === ObjectExplorerNodeType.Index) {
       this.currentObject = object
-      this.indexData = this.conversion.getIndexMapping(object.parent, this.conv, object.name)
+      this.indexData = this.conversion.getIndexMapping(object.parentId, this.conv, object.id)
     } else {
       this.currentObject = null
     }
diff --git a/ui/src/app/model/conv.ts b/ui/src/app/model/conv.ts
index 4eee7c9..981b4db 100644
--- a/ui/src/app/model/conv.ts
+++ b/ui/src/app/model/conv.ts
@@ -13,6 +13,18 @@ export default interface IConv {
   DatabaseType: string
   DatabaseName: string
   EditorName: string
+  Audit: IAudit
+}
+
+export interface IAudit {
+  ToSpannerFkIdx: Record<string, IFkeyAndIdxs>
+  ToSourceFkIdx: Record<string, IFkeyAndIdxs>
+}
+
+export interface IFkeyAndIdxs {
+  Name: string
+  ForeignKey: Record<string, string>
+  Index: Record<string, string>
 }
 
 export interface IStats {
@@ -31,6 +43,7 @@ export interface NameAndCols {
 // Spanner schema
 export interface ITable {
   Name: string
+  Id: string
   Schema: string
   ColNames: string[]
   ColDefs: Record<string, IColumn>
@@ -44,6 +57,7 @@ export interface IColumn {
   Type: ISpannerType
   NotNull: boolean
   Ignored: IIgnored
+  Id: string
 }
 
 export interface IIgnored {
@@ -65,6 +79,7 @@ export interface IIndex {
   Name: string
   Unique: boolean
   Keys: ISrcIndexKey[]
+  Id: string
 }
 
 export interface ISpannerForeignKey {
@@ -165,4 +180,4 @@ export interface ISessionSummary {
   SourceIndexCount: number
   SpannerIndexCount: number
   ConnectionType: string
-}
\ No newline at end of file
+}
diff --git a/ui/src/app/model/schema-object-node.ts b/ui/src/app/model/schema-object-node.ts
index f527f1a..373eabd 100644
--- a/ui/src/app/model/schema-object-node.ts
+++ b/ui/src/app/model/schema-object-node.ts
@@ -8,6 +8,9 @@ export default interface ISchemaObjectNode {
   parent: string
   pos: number
   isSpannerNode: boolean
+  isDeleted?: boolean
+  id: string
+  parentId: string
 }
 
 export interface FlatNode {
@@ -19,4 +22,7 @@ export interface FlatNode {
   pos: number
   level: number
   isSpannerNode: boolean
+  isDeleted: boolean
+  id: string
+  parentId: string
 }
diff --git a/ui/src/app/services/conversion/conversion.service.ts b/ui/src/app/services/conversion/conversion.service.ts
index e9f59dd..1dd2221 100644
--- a/ui/src/app/services/conversion/conversion.service.ts
+++ b/ui/src/app/services/conversion/conversion.service.ts
@@ -20,12 +20,23 @@ export class ConversionService {
     let spannerTableNames = Object.keys(conv.SpSchema).filter((name: string) =>
       name.toLocaleLowerCase().includes(searchText.toLocaleLowerCase())
     )
+    let spannerTableIds = spannerTableNames.map((name: string) => conv.SpSchema[name].Id)
+    let srcTableNames = Object.keys(conv.SrcSchema)
+
+    let deletedTableNames = srcTableNames.filter((srcTableName: string) => {
+      if (spannerTableIds.indexOf(conv.SrcSchema[srcTableName].Id) > -1) {
+        return false
+      }
+      return true
+    })
     let parentNode: ISchemaObjectNode = {
-      name: `Tables (${spannerTableNames.length})`,
+      name: `Tables (${srcTableNames.length})`,
       type: ObjectExplorerNodeType.Tables,
       parent: '',
       pos: -1,
       isSpannerNode: true,
+      id: '',
+      parentId: '',
       children: spannerTableNames.map((name: string) => {
         let spannerTable = conv.SpSchema[name]
         return {
@@ -35,6 +46,8 @@ export class ConversionService {
           parent: '',
           pos: -1,
           isSpannerNode: true,
+          id: spannerTable.Id,
+          parentId: '',
           children: [
             {
               name: `Indexes (${spannerTable.Indexes ? spannerTable.Indexes.length : 0})`,
@@ -43,6 +56,8 @@ export class ConversionService {
               parent: name,
               pos: -1,
               isSpannerNode: true,
+              id: '',
+              parentId: spannerTable.Id,
               children: spannerTable.Indexes
                 ? spannerTable.Indexes.map((index: ICreateIndex, i: number) => {
                     return {
@@ -51,6 +66,8 @@ export class ConversionService {
                       parent: name,
                       pos: i,
                       isSpannerNode: true,
+                      id: index.Id,
+                      parentId: spannerTable.Id,
                     }
                   })
                 : [],
@@ -65,6 +82,20 @@ export class ConversionService {
       parentNode.children?.sort((a, b) => (b.name > a.name ? 1 : a.name > b.name ? -1 : 0))
     }
 
+    deletedTableNames.forEach((tableName: string) => {
+      parentNode.children?.push({
+        name: tableName.replace(/[^A-Za-z0-9_]/g, '_'),
+        status: 'DARK',
+        type: ObjectExplorerNodeType.Table,
+        parent: '',
+        pos: -1,
+        isSpannerNode: true,
+        children: [],
+        isDeleted: true,
+        id: conv.SrcSchema[tableName].Id,
+        parentId: '',
+      })
+    })
     return [
       {
         name: conv.DatabaseName,
@@ -73,6 +104,8 @@ export class ConversionService {
         parent: '',
         pos: -1,
         isSpannerNode: true,
+        id: '',
+        parentId: '',
       },
     ]
   }
@@ -93,16 +126,20 @@ export class ConversionService {
       parent: '',
       pos: -1,
       isSpannerNode: false,
+      id: '',
+      parentId: '',
       children: srcTableNames.map((name: string) => {
         let srcTable = conv.SrcSchema[name]
-        let spname = conv.ToSpanner[name].Name
+        let spname = conv.ToSpanner[name] ? conv.ToSpanner[name].Name : ''
         return {
           name: name,
-          status: conversionRates[spname],
+          status: conversionRates[spname] ? conversionRates[spname] : 'NONE',
           type: ObjectExplorerNodeType.Table,
           parent: '',
           pos: -1,
           isSpannerNode: false,
+          id: srcTable.Id,
+          parentId: '',
           children: [
             {
               name: `Indexes (${srcTable.Indexes?.length || '0'})`,
@@ -111,6 +148,8 @@ export class ConversionService {
               parent: name,
               pos: -1,
               isSpannerNode: false,
+              id: '',
+              parentId: srcTable.Id,
               children: srcTable.Indexes
                 ? srcTable.Indexes.map((index: IIndex, i: number) => {
                     return {
@@ -119,6 +158,8 @@ export class ConversionService {
                       parent: name,
                       isSpannerNode: false,
                       pos: i,
+                      id: index.Id,
+                      parentId: srcTable.Id,
                     }
                   })
                 : [],
@@ -141,17 +182,25 @@ export class ConversionService {
         parent: '',
         isSpannerNode: false,
         pos: -1,
+        id: '',
+        parentId: '',
       },
     ]
   }
 
-  getColumnMapping(tableName: string, data: IConv): IColumnTabData[] {
-    let srcTableName = data.ToSource[tableName].Name
+  getColumnMapping(id: string, data: IConv): IColumnTabData[] {
+    let srcTableName = this.getSourceTableNameFromId(id, data)
+    let spTableName = this.getSpannerTableNameFromId(id, data)
+
+    let srcTableIds = Object.keys(data.SrcSchema[srcTableName].ColDefs).map(
+      (name: string) => data.SrcSchema[srcTableName].ColDefs[name].Id
+    )
+
     const res: IColumnTabData[] = data.SrcSchema[srcTableName].ColNames.map(
       (name: string, i: number) => {
-        let spColName = data.ToSpanner[srcTableName].Cols[name]
+        let spColName = data.ToSpanner[srcTableName]?.Cols[name]
         let srcPks = data.SrcSchema[srcTableName].PrimaryKeys
-        let spannerColDef = data.SpSchema[tableName].ColDefs[spColName]
+        let spannerColDef = spTableName ? data.SpSchema[spTableName]?.ColDefs[spColName] : null
         return {
           spOrder: spannerColDef ? i + 1 : '',
           srcOrder: i + 1,
@@ -159,34 +208,45 @@ export class ConversionService {
           spDataType: spannerColDef ? spannerColDef.T.Name : '',
           srcColName: name,
           srcDataType: data.SrcSchema[srcTableName].ColDefs[name].Type.Name,
-          spIsPk: spannerColDef
-            ? data.SpSchema[tableName].Pks.map((p) => p.Col).indexOf(spColName) !== -1
-            : false,
+          spIsPk:
+            spannerColDef && spTableName
+              ? data.SpSchema[spTableName].Pks?.map((p) => p.Col).indexOf(spColName) !== -1
+              : false,
           srcIsPk: srcPks ? srcPks.map((p) => p.Column).indexOf(name) !== -1 : false,
-          spIsNotNull: spannerColDef ? data.SpSchema[tableName].ColDefs[spColName].NotNull : false,
+          spIsNotNull:
+            spannerColDef && spTableName
+              ? data.SpSchema[spTableName].ColDefs[spColName].NotNull
+              : false,
           srcIsNotNull: data.SrcSchema[srcTableName].ColDefs[name].NotNull,
         }
       }
     )
-    data.SpSchema[srcTableName].ColNames.forEach((name: string, i: number) => {
-      if (data.SrcSchema[srcTableName].ColNames.indexOf(name) < 0) {
-        let spannerColDef = data.SpSchema[tableName].ColDefs[name]
-        res.push({
-          spOrder: i + 1,
-          srcOrder: '',
-          spColName: name,
-          spDataType: spannerColDef ? spannerColDef.T.Name : '',
-          srcColName: '',
-          srcDataType: '',
-          spIsPk: spannerColDef
-            ? data.SpSchema[tableName].Pks.map((p) => p.Col).indexOf(name) !== -1
-            : false,
-          srcIsPk: false,
-          spIsNotNull: spannerColDef ? data.SpSchema[tableName].ColDefs[name].NotNull : false,
-          srcIsNotNull: false,
-        })
-      }
-    })
+    if (spTableName) {
+      data.SpSchema[spTableName]?.ColNames.forEach((name: string, i: number) => {
+        if (spTableName && srcTableIds.indexOf(data.SpSchema[spTableName].ColDefs[name].Id) < 0) {
+          let spannerColDef = spTableName ? data.SpSchema[spTableName].ColDefs[name] : null
+          res.push({
+            spOrder: i + 1,
+            srcOrder: '',
+            spColName: name,
+            spDataType: spannerColDef ? spannerColDef.T.Name : '',
+            srcColName: '',
+            srcDataType: '',
+            spIsPk:
+              spannerColDef && spTableName
+                ? data.SpSchema[spTableName].Pks.map((p) => p.Col).indexOf(name) !== -1
+                : false,
+            srcIsPk: false,
+            spIsNotNull:
+              spannerColDef && spTableName
+                ? data.SpSchema[spTableName].ColDefs[name].NotNull
+                : false,
+            srcIsNotNull: false,
+          })
+        }
+      })
+    }
+
     return res
   }
 
@@ -197,17 +257,26 @@ export class ConversionService {
     return JSON.parse(JSON.stringify(pkColumns))
   }
 
-  getFkMapping(tableName: string, data: IConv): IFkTabData[] {
-    let srcTableName: string = data.ToSource[tableName].Name
-    let spFks = data.SpSchema[tableName].Fks
-    if (!spFks) {
+  getFkMapping(id: string, data: IConv): IFkTabData[] {
+    let srcTableName = this.getSourceTableNameFromId(id, data)
+    let spTableName = this.getSpannerTableNameFromId(id, data)
+
+    let spFks =
+      spTableName && data.SpSchema[spTableName] && data.SpSchema[spTableName].Fks
+        ? data.SpSchema[spTableName].Fks
+        : []
+    let srcFks = data.SrcSchema[srcTableName]?.ForeignKeys
+
+    if (!srcFks) {
       return []
     }
     let spFklength: number = spFks.length
     return data.SrcSchema[srcTableName].ForeignKeys.map((item: ISpannerForeignKey, i: number) => {
       spFklength = spFklength - 1
       if (
-        data.SpSchema[tableName].Fks.length != data.SrcSchema[srcTableName].ForeignKeys.length &&
+        spTableName &&
+        data.SpSchema[spTableName] &&
+        data.SpSchema[spTableName].Fks?.length != data.SrcSchema[srcTableName].ForeignKeys.length &&
         spFklength < 0
       ) {
         return {
@@ -222,43 +291,60 @@ export class ConversionService {
         }
       } else {
         return {
-          spName: data.SpSchema[tableName].Fks[i].Name,
+          spName:
+            spTableName && data.SpSchema[spTableName] ? data.SpSchema[spTableName].Fks[i].Name : '',
           srcName: data.SrcSchema[srcTableName].ForeignKeys[i].Name,
-          spColumns: data.SpSchema[tableName].Fks[i].Columns,
+          spColumns:
+            spTableName && data.SpSchema[spTableName]
+              ? data.SpSchema[spTableName].Fks[i].Columns
+              : [],
           srcColumns: data.SrcSchema[srcTableName].ForeignKeys[i].Columns,
-          spReferTable: data.SpSchema[tableName].Fks[i].ReferTable,
+          spReferTable:
+            spTableName && data.SpSchema[spTableName]
+              ? data.SpSchema[spTableName].Fks[i].ReferTable
+              : '',
           srcReferTable: data.SrcSchema[srcTableName].ForeignKeys[i].ReferTable,
-          spReferColumns: data.SpSchema[tableName].Fks[i].ReferColumns,
+          spReferColumns:
+            spTableName && data.SpSchema[spTableName]
+              ? data.SpSchema[spTableName].Fks[i].ReferColumns
+              : [],
           srcReferColumns: data.SrcSchema[srcTableName].ForeignKeys[i].ReferColumns,
         }
       }
     })
   }
 
-  getIndexMapping(tableName: string, data: IConv, indexName: string): IIndexData[] {
-    let srcTableName = data.ToSource[tableName].Name
-    let spIndex = data.SpSchema[tableName].Indexes.filter((idx) => idx.Name === indexName)[0]
-    let srcIndexs = data.SrcSchema[srcTableName].Indexes?.filter((idx) => idx.Name === indexName)
+  getIndexMapping(tableId: string, data: IConv, indexId: string): IIndexData[] {
+    let srcTableName = this.getSourceTableNameFromId(tableId, data)
+    let spTableName = this.getSpannerTableNameFromId(tableId, data)
+    let spIndex = spTableName
+      ? data.SpSchema[spTableName]?.Indexes.filter((idx) => idx.Id === indexId)[0]
+      : null
+    let srcIndexs = data.SrcSchema[srcTableName].Indexes?.filter((idx) => idx.Id === indexId)
 
-    let res: IIndexData[] = spIndex.Keys.map((idx: IIndexKey, i: number) => {
-      return {
-        srcColName:
-          srcIndexs && srcIndexs.length > 0 && srcIndexs[0].Keys.length > i
-            ? srcIndexs[0].Keys[i].Column
-            : '',
-        srcOrder: srcIndexs && srcIndexs.length > 0 && srcIndexs[0].Keys.length > i ? i + 1 : '',
-        srcDesc:
-          srcIndexs && srcIndexs.length > 0 && srcIndexs[0].Keys.length > i
-            ? srcIndexs[0].Keys[i].Desc
-            : undefined,
-        spColName: idx.Col,
-        spOrder: i + 1,
-        spDesc: idx.Desc,
-      }
-    })
-    if (srcIndexs && srcIndexs[0] && spIndex.Keys.length < srcIndexs[0].Keys.length) {
+    let res: IIndexData[] = spIndex
+      ? spIndex.Keys.map((idx: IIndexKey, i: number) => {
+          return {
+            srcColName:
+              srcIndexs && srcIndexs.length > 0 && srcIndexs[0].Keys.length > i
+                ? srcIndexs[0].Keys[i].Column
+                : '',
+            srcOrder:
+              srcIndexs && srcIndexs.length > 0 && srcIndexs[0].Keys.length > i ? i + 1 : '',
+            srcDesc:
+              srcIndexs && srcIndexs.length > 0 && srcIndexs[0].Keys.length > i
+                ? srcIndexs[0].Keys[i].Desc
+                : undefined,
+            spColName: idx.Col,
+            spOrder: i + 1,
+            spDesc: idx.Desc,
+          }
+        })
+      : []
+    let spKeyLength = spIndex ? spIndex.Keys.length : 0
+    if (srcIndexs && srcIndexs[0] && spKeyLength < srcIndexs[0].Keys.length) {
       srcIndexs[0].Keys.forEach((idx, index) => {
-        if (index >= spIndex.Keys.length) {
+        if (index >= spKeyLength) {
           res.push({
             srcColName: idx.Column,
             srcOrder: index + 1,
@@ -272,4 +358,23 @@ export class ConversionService {
     }
     return res
   }
+
+  getSourceTableNameFromId(id: string, conv: IConv): string {
+    let srcName: string = ''
+    Object.keys(conv.SrcSchema).forEach((key: string) => {
+      if (conv.SrcSchema[key].Id === id) {
+        srcName = conv.SrcSchema[key].Name
+      }
+    })
+    return srcName
+  }
+  getSpannerTableNameFromId(id: string, conv: IConv): string | null {
+    let spName: string | null = null
+    Object.keys(conv.SpSchema).forEach((key: string) => {
+      if (conv.SpSchema[key].Id === id) {
+        spName = conv.SpSchema[key].Name
+      }
+    })
+    return spName
+  }
 }
diff --git a/ui/src/app/services/data/data.service.ts b/ui/src/app/services/data/data.service.ts
index 9fee7f6..1691d6e 100644
--- a/ui/src/app/services/data/data.service.ts
+++ b/ui/src/app/services/data/data.service.ts
@@ -197,6 +197,44 @@ export class DataService {
     )
   }
 
+  restoreTable(tableId: string): Observable<string> {
+    return this.fetch.restoreTable(tableId).pipe(
+      catchError((e: any) => {
+        return of({ error: e.error })
+      }),
+      tap(console.log),
+      map((data) => {
+        if (data.error) {
+          this.snackbar.openSnackBar(data.error, 'Close')
+          return data.error
+        } else {
+          this.convSubject.next(data)
+          this.snackbar.openSnackBar('Table restored successfully', 'Close', 5)
+          return ''
+        }
+      })
+    )
+  }
+
+  dropTable(tableId: string): Observable<string> {
+    return this.fetch.dropTable(tableId).pipe(
+      catchError((e: any) => {
+        return of({ error: e.error })
+      }),
+      tap(console.log),
+      map((data) => {
+        if (data.error) {
+          this.snackbar.openSnackBar(data.error, 'Close')
+          return data.error
+        } else {
+          this.convSubject.next(data)
+          this.snackbar.openSnackBar('Table dropped successfully', 'Close', 5)
+          return ''
+        }
+      })
+    )
+  }
+
   updatePk(pkObj: IPrimaryKey) {
     return this.fetch.updatePk(pkObj).pipe(
       catchError((e: any) => {
diff --git a/ui/src/app/services/fetch/fetch.service.ts b/ui/src/app/services/fetch/fetch.service.ts
index 12b15f4..9ee972c 100644
--- a/ui/src/app/services/fetch/fetch.service.ts
+++ b/ui/src/app/services/fetch/fetch.service.ts
@@ -65,6 +65,13 @@ export class FetchService {
     return this.http.post<HttpResponse<IConv>>(`${this.url}/typemap/table?table=${tableName}`, data)
   }
 
+  restoreTable(tableId: string) {
+    return this.http.post<HttpResponse<IConv>>(`${this.url}/restore/table?tableId=${tableId}`, {})
+  }
+  dropTable(tableId: string) {
+    return this.http.post<HttpResponse<IConv>>(`${this.url}/drop/table?tableId=${tableId}`, {})
+  }
+
   updatePk(pkObj: IPrimaryKey) {
     return this.http.post<HttpResponse<IConv>>(`${this.url}/primaryKey`, pkObj)
   }
diff --git a/ui/src/styles.scss b/ui/src/styles.scss
index f17fc50..9b7ce9b 100644
--- a/ui/src/styles.scss
+++ b/ui/src/styles.scss
@@ -44,6 +44,28 @@ body {
   color: #3367d6;
 }
 
+.gray-out {
+  opacity: 0.6;
+}
+
+.danger {
+  color: #da4236;
+}
+
+.warning {
+  color: #f3b300;
+}
+
+.success {
+  color: green;
+}
+
+.orange {
+  color: rgb(248, 116, 68);
+}
+.dark {
+  color: rgb(77, 75, 75);
+}
 .header {
   padding: 0 20px;
   display: flex;
diff --git a/webv2/index/index.go b/webv2/index/index.go
index 3698e09..af7f849 100644
--- a/webv2/index/index.go
+++ b/webv2/index/index.go
@@ -59,7 +59,7 @@ func checkRedundantIndex(index []ddl.CreateIndex, spannerTable ddl.CreateTable)
 		indexFirstColumn := index[i].Keys[0].Col
 
 		if primaryKeyFirstColumn == indexFirstColumn {
-			columnname := keys[i].Col
+			columnname := keys[0].Col
 			sessionState := session.GetSessionState()
 			schemaissue := sessionState.Conv.Issues[spannerTable.Name][columnname]
 			schemaissue = append(schemaissue, internal.RedundantIndex)
diff --git a/webv2/routes.go b/webv2/routes.go
index e617da4..d7b25c0 100644
--- a/webv2/routes.go
+++ b/webv2/routes.go
@@ -45,6 +45,9 @@ func getRoutes() *mux.Router {
 	// TODO:(searce) take constraint names themselves which are guaranteed to be unique for Spanner.
 	router.HandleFunc("/drop/secondaryindex", dropSecondaryIndex).Methods("POST")
 
+	router.HandleFunc("/restore/table", restoreTable).Methods("POST")
+	router.HandleFunc("/drop/table", dropTable).Methods("POST")
+
 	router.HandleFunc("/rename/fks", renameForeignKeys).Methods("POST")
 	router.HandleFunc("/rename/indexes", renameIndexes).Methods("POST")
 	router.HandleFunc("/add/indexes", addIndexes).Methods("POST")
diff --git a/webv2/session/remote_session_store.go b/webv2/session/remote_session_store.go
index 51a3cbd..9884be6 100644
--- a/webv2/session/remote_session_store.go
+++ b/webv2/session/remote_session_store.go
@@ -112,9 +112,8 @@ func (st *spannerStore) GetConvWithMetadata(ctx context.Context, versionId strin
 	}
 
 	convm.Conv = conv
-	convm.Conv.Audit = internal.Audit{
-		MigrationType: migration.MigrationData_SCHEMA_ONLY.Enum(),
-	}
+	convm.Conv.Audit.MigrationType = migration.MigrationData_SCHEMA_ONLY.Enum()
+
 	convm.SessionMetadata = SessionMetadata{
 		SessionName:  scs.SessionName,
 		EditorName:   scs.EditorName,
diff --git a/webv2/uniqueid/uniqueid_assignment.go b/webv2/uniqueid/uniqueid_assignment.go
index 5f24b86..bd6db08 100644
--- a/webv2/uniqueid/uniqueid_assignment.go
+++ b/webv2/uniqueid/uniqueid_assignment.go
@@ -32,7 +32,7 @@ func AssignUniqueId(conv *internal.Conv) {
 
 		for spannertablename, spannertable := range conv.SpSchema {
 
-			if sourcetablename == spannertablename {
+			if validSpannerTableName, _ := internal.FixName(sourcetablename); validSpannerTableName == spannertablename {
 
 				tableuniqueid := GenerateTableId()
 				sourcetable.Id = tableuniqueid
@@ -53,7 +53,7 @@ func AssignUniqueId(conv *internal.Conv) {
 
 					for spannercolumnname, spannercolumn := range spannertable.ColDefs {
 
-						if sourcecolumn.Name == spannercolumn.Name {
+						if validSpannerColumnName, _ := internal.FixName(sourcecolumn.Name); validSpannerColumnName == spannercolumn.Name {
 
 							columnuniqueid := GenerateColumnId()
 							sourcecolumn.Id = columnuniqueid
@@ -72,7 +72,7 @@ func AssignUniqueId(conv *internal.Conv) {
 
 					for spannerforeignkeyindex, spannerforeignkey := range spannertable.Fks {
 
-						if sourceforeignkey.Name == spannerforeignkey.Name {
+						if validSpannerFkName, _ := internal.FixName(sourceforeignkey.Name); validSpannerFkName == spannerforeignkey.Name {
 
 							foreignkeyid := GenerateForeignkeyId()
 
@@ -88,9 +88,9 @@ func AssignUniqueId(conv *internal.Conv) {
 
 				for sourcei, sourceindexes := range sourcetable.Indexes {
 
-					for spanneri, spannerindexes := range spannertable.Indexes {
+					for ind, spannerindexes := range spannertable.Indexes {
 
-						if sourceindexes.Name == spannerindexes.Name {
+						if validSpannerIndexName, _ := internal.FixName(sourceindexes.Name); validSpannerIndexName == spannerindexes.Name {
 
 							indexesid := GenerateIndexesId()
 
@@ -98,7 +98,7 @@ func AssignUniqueId(conv *internal.Conv) {
 							spannerindexes.Id = indexesid
 
 							conv.SrcSchema[sourcetable.Name].Indexes[sourcei] = sourceindexes
-							conv.SpSchema[spannertable.Name].Indexes[spanneri] = spannerindexes
+							conv.SpSchema[spannertable.Name].Indexes[ind] = spannerindexes
 
 						}
 
@@ -156,7 +156,7 @@ func updateSpannerTableSecondaryIndexKeyOrder(spannertable ddl.CreateTable) {
 
 		for j := 0; j < len(spannertable.Indexes[i].Keys); j++ {
 
-			for spannercolumnname, _ := range spannertable.ColDefs {
+			for spannercolumnname := range spannertable.ColDefs {
 				if spannertable.Indexes[i].Keys[j].Col == spannercolumnname {
 
 					o := getSpannerColumnIndex(spannertable, spannercolumnname)
@@ -175,7 +175,7 @@ func updateSourceTableSecondaryIndexKeyOrder(sourcetable schema.Table) {
 
 		for j := 0; j < len(sourcetable.Indexes[i].Keys); j++ {
 
-			for spannercolumnname, _ := range sourcetable.ColDefs {
+			for spannercolumnname := range sourcetable.ColDefs {
 				if sourcetable.Indexes[i].Keys[j].Column == spannercolumnname {
 
 					o := getSourceColumnIndex(sourcetable, spannercolumnname)
@@ -258,3 +258,50 @@ func InitObjectId() {
 	sessionState := session.GetSessionState()
 	sessionState.Counter.ObjectId = "0"
 }
+
+//CopyUniqueIdToSpannerTable copy ids from source table to spanner table content
+func CopyUniqueIdToSpannerTable(conv *internal.Conv, spannertablename string) {
+	sourcetablename := conv.ToSource[spannertablename].Name
+	spannertable := conv.SpSchema[spannertablename]
+	sourcetable := conv.SrcSchema[sourcetablename]
+	spannertable.Id = sourcetable.Id
+
+	for spannercolumnname, spannercolumn := range spannertable.ColDefs {
+		if spannercolumn.Name == "synth_id" {
+			columnuniqueid := GenerateColumnId()
+			spannercolumn.Id = columnuniqueid
+			spannertable.ColDefs[spannercolumnname] = spannercolumn
+		}
+	}
+
+	for _, sourcecolumn := range sourcetable.ColDefs {
+		for spannercolumnname, spannercolumn := range spannertable.ColDefs {
+			if validSpannerColumnName, _ := internal.FixName(sourcecolumn.Name); validSpannerColumnName == spannercolumn.Name {
+				spannercolumn.Id = sourcecolumn.Id
+				spannertable.ColDefs[spannercolumnname] = spannercolumn
+				break
+			}
+		}
+	}
+
+	for _, sourceforeignkey := range sourcetable.ForeignKeys {
+		for spannerforeignkeyindex, spannerforeignkey := range spannertable.Fks {
+			if validSpannerFkName, _ := internal.FixName(sourceforeignkey.Name); validSpannerFkName == spannerforeignkey.Name {
+				spannerforeignkey.Id = sourceforeignkey.Id
+				spannertable.Fks[spannerforeignkeyindex] = spannerforeignkey
+			}
+		}
+	}
+
+	for _, sourceindexes := range sourcetable.Indexes {
+		for ind, spannerindexes := range spannertable.Indexes {
+			if validSpannerIndexName, _ := internal.FixName(sourceindexes.Name); validSpannerIndexName == spannerindexes.Name {
+				spannerindexes.Id = sourceindexes.Id
+				spannertable.Indexes[ind] = spannerindexes
+			}
+		}
+	}
+
+	updateSpannerTableIndexKeyOrder(spannertable)
+	conv.SpSchema[spannertablename] = spannertable
+}
diff --git a/webv2/web.go b/webv2/web.go
index 6fab46a..8ddebc5 100644
--- a/webv2/web.go
+++ b/webv2/web.go
@@ -772,6 +772,174 @@ type DropDetail struct {
 	Name string `json:"Name"`
 }
 
+func restoreTable(w http.ResponseWriter, r *http.Request) {
+	tableId := r.FormValue("tableId")
+	sessionState := session.GetSessionState()
+	if sessionState.Conv == nil || sessionState.Driver == "" {
+		http.Error(w, fmt.Sprintf("Schema is not converted or Driver is not configured properly. Please retry converting the database to Spanner."), http.StatusNotFound)
+		return
+	}
+	if tableId == "" {
+		http.Error(w, fmt.Sprintf("Table Id is empty"), http.StatusBadRequest)
+	}
+
+	table := ""
+	for _, value := range sessionState.Conv.SrcSchema {
+		if value.Id == tableId {
+			table = value.Name
+			break
+		}
+	}
+	if table == "" {
+		http.Error(w, fmt.Sprintf("Table not found"), http.StatusBadRequest)
+	}
+
+	conv := sessionState.Conv
+	var toddl common.ToDdl
+	switch sessionState.Driver {
+	case constants.MYSQL:
+		toddl = mysql.InfoSchemaImpl{}.GetToDdl()
+	case constants.POSTGRES:
+		toddl = postgres.InfoSchemaImpl{}.GetToDdl()
+	case constants.SQLSERVER:
+		toddl = sqlserver.InfoSchemaImpl{}.GetToDdl()
+	case constants.ORACLE:
+		toddl = oracle.InfoSchemaImpl{}.GetToDdl()
+	case constants.MYSQLDUMP:
+		toddl = mysql.DbDumpImpl{}.GetToDdl()
+	case constants.PGDUMP:
+		toddl = postgres.DbDumpImpl{}.GetToDdl()
+	default:
+		http.Error(w, fmt.Sprintf("Driver : '%s' is not supported", sessionState.Driver), http.StatusBadRequest)
+		return
+	}
+
+	err := common.SrcTableToSpannerDDL(conv, toddl, sessionState.Conv.SrcSchema[table])
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Restoring spanner table fail"), http.StatusBadRequest)
+		return
+	}
+	conv.AddPrimaryKeys()
+	for _, spTable := range conv.SpSchema {
+		uniqueid.CopyUniqueIdToSpannerTable(conv, spTable.Name)
+	}
+	sessionState.Conv = conv
+	primarykey.DetectHotspot()
+
+	convm := session.ConvWithMetadata{
+		SessionMetadata: sessionState.SessionMetadata,
+		Conv:            *sessionState.Conv,
+	}
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(convm)
+}
+
+func dropTable(w http.ResponseWriter, r *http.Request) {
+	tableId := r.FormValue("tableId")
+	sessionState := session.GetSessionState()
+	if sessionState.Conv == nil || sessionState.Driver == "" {
+		http.Error(w, fmt.Sprintf("Schema is not converted or Driver is not configured properly. Please retry converting the database to Spanner."), http.StatusNotFound)
+		return
+	}
+	if tableId == "" {
+		http.Error(w, fmt.Sprintf("Table Id is empty"), http.StatusBadRequest)
+	}
+	table := ""
+	for _, value := range sessionState.Conv.SpSchema {
+		if value.Id == tableId {
+			table = value.Name
+			break
+		}
+	}
+	if table == "" {
+		http.Error(w, fmt.Sprintf("Spanner table not found"), http.StatusBadRequest)
+	}
+
+	srcTable := ""
+	for _, value := range sessionState.Conv.SrcSchema {
+		if value.Id == tableId {
+			srcTable = value.Name
+			break
+		}
+	}
+	if srcTable == "" {
+		http.Error(w, fmt.Sprintf("Source table not found"), http.StatusBadRequest)
+	}
+
+	spSchema := sessionState.Conv.SpSchema
+	toSource := sessionState.Conv.ToSource
+	toSpanner := sessionState.Conv.ToSpanner
+	issues := sessionState.Conv.Issues
+	syntheticPkey := sessionState.Conv.SyntheticPKeys
+	toSourceFkIdx := sessionState.Conv.Audit.ToSourceFkIdx
+	toSpannerFkIdx := sessionState.Conv.Audit.ToSpannerFkIdx
+
+	//remove deleted name from usedName
+	usedNames := sessionState.Conv.UsedNames
+	delete(usedNames, table)
+	for _, index := range sessionState.Conv.SpSchema[table].Indexes {
+		delete(usedNames, index.Name)
+	}
+	for _, fk := range sessionState.Conv.SpSchema[table].Fks {
+		delete(usedNames, fk.Name)
+	}
+
+	delete(spSchema, table)
+	delete(toSource, table)
+	delete(toSpanner, srcTable)
+	issues[srcTable] = map[string][]internal.SchemaIssue{}
+	delete(syntheticPkey, table)
+	delete(toSourceFkIdx, table)
+	delete(toSpannerFkIdx, srcTable)
+
+	//drop reference foreign key
+	for tableName, spTable := range spSchema {
+		fks := []ddl.Foreignkey{}
+		for _, fk := range spTable.Fks {
+			if fk.ReferTable != table {
+				fks = append(fks, fk)
+			} else {
+				delete(usedNames, fk.Name)
+			}
+
+		}
+		spTable.Fks = fks
+		spSchema[tableName] = spTable
+	}
+
+	//remove interleave that are interleaved on the drop table as parent
+	for tableName, spTable := range spSchema {
+		if spTable.Parent == table {
+			spTable.Parent = ""
+			spSchema[tableName] = spTable
+		}
+	}
+
+	//remove interleavable suggestion on droping the parent table
+	for tableName, tableIssues := range issues {
+		for colName, colIssues := range tableIssues {
+			updatedColIssues := []internal.SchemaIssue{}
+			for _, val := range colIssues {
+				if val != internal.InterleavedOrder {
+					updatedColIssues = append(updatedColIssues, val)
+				}
+			}
+			if len(updatedColIssues) == 0 {
+				delete(issues[tableName], colName)
+			} else {
+				issues[tableName][colName] = updatedColIssues
+			}
+		}
+	}
+
+	convm := session.ConvWithMetadata{
+		SessionMetadata: sessionState.SessionMetadata,
+		Conv:            *sessionState.Conv,
+	}
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(convm)
+}
+
 func dropForeignKey(w http.ResponseWriter, r *http.Request) {
 	table := r.FormValue("table")
 	reqBody, err := ioutil.ReadAll(r.Body)
@@ -1468,15 +1636,12 @@ func checkPrimaryKeyPrefix(table string, refTable string, fk ddl.Foreignkey, tab
 
 			if len(schemaissue) > 0 {
 
-				if sessionState.Conv.Issues[table][caninterleaved[i]] == nil {
+				if sessionState.Conv.Issues[table] == nil {
 
 					s := map[string][]internal.SchemaIssue{
 						caninterleaved[i]: schemaissue,
 					}
-					sessionState.Conv.Issues = map[string]map[string][]internal.SchemaIssue{}
-
 					sessionState.Conv.Issues[table] = s
-
 				} else {
 					sessionState.Conv.Issues[table][caninterleaved[i]] = schemaissue
 				}


Test patch:
diff --git a/webv2/web_test.go b/webv2/web_test.go
index 23c556f..d71652d 100644
--- a/webv2/web_test.go
+++ b/webv2/web_test.go
@@ -1101,6 +1101,7 @@ func TestSetParentTable(t *testing.T) {
 			name:       "no table name provided",
 			statusCode: http.StatusBadRequest,
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{"t1": {
 					Name:     "t1",
 					ColNames: []string{"a", "b", "c"},
@@ -1119,6 +1120,7 @@ func TestSetParentTable(t *testing.T) {
 		{
 			name: "table with synthetic PK",
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{"t1": {
 					Name:     "t1",
 					ColNames: []string{"a", "b", "c"},
@@ -1143,6 +1145,7 @@ func TestSetParentTable(t *testing.T) {
 		{
 			name: "no valid prefix 1",
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{
 					"t1": {
 						Name:     "t1",
@@ -1178,6 +1181,7 @@ func TestSetParentTable(t *testing.T) {
 		{
 			name: "no valid prefix 2",
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{
 					"t1": {
 						Name:     "t1",
@@ -1211,6 +1215,7 @@ func TestSetParentTable(t *testing.T) {
 		{
 			name: "no valid prefix 3",
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{
 					"t1": {
 						Name:     "t1",
@@ -1244,6 +1249,7 @@ func TestSetParentTable(t *testing.T) {
 		{
 			name: "successful interleave",
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{
 					"t1": {
 						Name:     "t1",
@@ -1278,6 +1284,7 @@ func TestSetParentTable(t *testing.T) {
 		{
 			name: "successful interleave with same primary key",
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{
 					"t1": {
 						Name:     "t1",
@@ -1312,6 +1319,7 @@ func TestSetParentTable(t *testing.T) {
 		{
 			name: "successful interleave with multiple fks refering multiple tables",
 			ct: &internal.Conv{
+				Issues: map[string]map[string][]internal.SchemaIssue{},
 				SpSchema: map[string]ddl.CreateTable{
 					"t1": {
 						Name:     "t1",
@@ -2388,6 +2396,231 @@ func TestDropSecondaryIndex(t *testing.T) {
 	}
 }
 
+func TestDropTable(t *testing.T) {
+	sessionState := session.GetSessionState()
+	sessionState.Driver = constants.MYSQL
+
+	c := &internal.Conv{
+		Issues: map[string]map[string][]internal.SchemaIssue{
+			"t1": {},
+			"t2": {},
+		},
+		SrcSchema: map[string]schema.Table{
+			"t1": {
+				Name:     "t1",
+				ColNames: []string{"a", "b", "c"},
+				ColDefs: map[string]schema.Column{
+					"a": {Name: "a", Type: schema.Type{Name: "bigint"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c4"},
+					"b": {Name: "b", Type: schema.Type{Name: "bigint"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c2"},
+					"c": {Name: "c", Type: schema.Type{Name: "varchar"}, NotNull: false, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c3"},
+				},
+				PrimaryKeys: []schema.Key{{Column: "a", Desc: false, Order: 1}},
+				Id:          "id1",
+			},
+
+			"t2": {
+				Name:     "t2",
+				ColNames: []string{"a", "b", "c"},
+				ColDefs: map[string]schema.Column{
+					"a": {Name: "a", Type: schema.Type{Name: "bigint"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c7"},
+					"b": {Name: "b", Type: schema.Type{Name: "varchar"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c8"},
+					"c": {Name: "c", Type: schema.Type{Name: "bigint"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c9"},
+				},
+				Id: "id2",
+			},
+		},
+		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}},
+				Id:  "id1",
+			},
+			"t2": {
+				Name:     "t2",
+				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},
+					"synth_id": {Name: "synth_id", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+				},
+				Pks: []ddl.IndexKey{{Col: "synth_id", Desc: false}},
+				Id:  "id2",
+			}},
+		Audit: internal.Audit{
+			MigrationType: migration.MigrationData_MIGRATION_TYPE_UNSPECIFIED.Enum(),
+		},
+	}
+
+	sessionState.Conv = c
+
+	payload := `{}`
+
+	req, err := http.NewRequest("POST", "/drop/table?tableId=id1", strings.NewReader(payload))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	rr := httptest.NewRecorder()
+
+	handler := http.HandlerFunc(dropTable)
+	handler.ServeHTTP(rr, req)
+
+	res := &internal.Conv{}
+
+	json.Unmarshal(rr.Body.Bytes(), &res)
+
+	expectedConv := &internal.Conv{
+
+		SpSchema: map[string]ddl.CreateTable{
+			"t2": {
+				Name:     "t2",
+				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},
+					"synth_id": {Name: "synth_id", T: ddl.Type{Name: ddl.Int64}, NotNull: true},
+				},
+				Pks:     []ddl.IndexKey{{Col: "synth_id", Desc: false}},
+				Fks:     []ddl.Foreignkey{},
+				Indexes: []ddl.CreateIndex(nil),
+				Id:      "id2",
+			}},
+	}
+
+	assert.Equal(t, expectedConv.SpSchema, res.SpSchema)
+}
+
+func TestRestoreTable(t *testing.T) {
+	sessionState := session.GetSessionState()
+
+	sessionState.Driver = constants.MYSQL
+
+	c := &internal.Conv{
+
+		SpSchema: map[string]ddl.CreateTable{
+			"t2": {
+				Name:     "t2",
+				ColNames: []string{"a", "b"},
+				ColDefs: map[string]ddl.ColumnDef{"a": {Name: "a", T: ddl.Type{Name: "STRING", IsArray: false}, NotNull: true, Comment: "", Id: "c7"},
+					"b": {Name: "b", T: ddl.Type{Name: "STRING", IsArray: false}, NotNull: true, Comment: "", Id: "c8"},
+				},
+				Pks: []ddl.IndexKey{{Col: "a", Desc: false, Order: 1}},
+				Id:  "t6",
+			}},
+		SrcSchema: map[string]schema.Table{
+			"t1": {
+				Name:     "t1",
+				ColNames: []string{"a", "x", "y"},
+				ColDefs: map[string]schema.Column{
+					"a": {Name: "a", Type: schema.Type{Name: "varchar"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c4"},
+					"x": {Name: "x", Type: schema.Type{Name: "varchar"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c2"},
+					"y": {Name: "y", Type: schema.Type{Name: "bigint"}, NotNull: false, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c3"},
+				},
+				PrimaryKeys: []schema.Key{{Column: "a", Desc: false, Order: 1}},
+				Id:          "t1",
+			},
+
+			"t2": {
+				Name:     "t2",
+				ColNames: []string{"a", "b"},
+				ColDefs: map[string]schema.Column{
+					"a": {Name: "a", Type: schema.Type{Name: "varchar"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c7"},
+					"b": {Name: "b", Type: schema.Type{Name: "char"}, NotNull: true, Ignored: schema.Ignored{Check: false, Identity: false, Default: false, Exclusion: false, ForeignKey: false, AutoIncrement: false}, Id: "c8"},
+				},
+				PrimaryKeys: []schema.Key{{Column: "a", Desc: false, Order: 1}},
+				Id:          "t6",
+			},
+		},
+
+		UsedNames: map[string]bool{
+			"t2": true,
+		},
+
+		ToSource: map[string]internal.NameAndCols{
+			"t2": {Name: "t2", Cols: map[string]string{"a": "a", "b": "b"}},
+		},
+		ToSpanner: map[string]internal.NameAndCols{
+			"t2": {Name: "t2", Cols: map[string]string{"a": "a", "b": "b"}},
+		},
+
+		Audit: internal.Audit{
+			MigrationType: migration.MigrationData_MIGRATION_TYPE_UNSPECIFIED.Enum(),
+			ToSourceFkIdx: map[string]internal.FkeyAndIdxs{
+				"t2": {
+					Name:       "t2",
+					ForeignKey: map[string]string{},
+					Index:      map[string]string{},
+				},
+			},
+			ToSpannerFkIdx: map[string]internal.FkeyAndIdxs{
+				"t2": {
+					Name:       "t2",
+					ForeignKey: map[string]string{},
+					Index:      map[string]string{},
+				},
+			},
+		},
+
+		Issues: map[string]map[string][]internal.SchemaIssue{},
+	}
+
+	sessionState.Conv = c
+
+	payload := `{}`
+
+	req, err := http.NewRequest("POST", "/restore/table?tableId=t1", strings.NewReader(payload))
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	rr := httptest.NewRecorder()
+
+	handler := http.HandlerFunc(restoreTable)
+	handler.ServeHTTP(rr, req)
+
+	res := &internal.Conv{}
+
+	json.Unmarshal(rr.Body.Bytes(), &res)
+
+	expectedConv := &internal.Conv{
+
+		SpSchema: map[string]ddl.CreateTable{
+
+			"t1": {
+				Name:     "t1",
+				ColNames: []string{"a", "x", "y"},
+				ColDefs: map[string]ddl.ColumnDef{
+					"a": {Name: "a", T: ddl.Type{Name: "STRING", Len: 9223372036854775807, IsArray: false}, NotNull: true, Comment: "From: a varchar", Id: "c4"},
+					"x": {Name: "x", T: ddl.Type{Name: "STRING", Len: 9223372036854775807, IsArray: false}, NotNull: true, Comment: "From: x varchar", Id: "c2"},
+					"y": {Name: "y", T: ddl.Type{Name: "INT64", Len: 0, IsArray: false}, NotNull: false, Comment: "From: y bigint", Id: "c3"},
+				},
+				Pks:     []ddl.IndexKey{{Col: "a", Desc: false, Order: 1}},
+				Id:      "t1",
+				Comment: "Spanner schema for source table t1",
+			},
+
+			"t2": {
+				Name:     "t2",
+				ColNames: []string{"a", "b"},
+				ColDefs: map[string]ddl.ColumnDef{"a": {Name: "a", T: ddl.Type{Name: "STRING", Len: 0, IsArray: false}, NotNull: true, Comment: "", Id: "c7"},
+					"b": {Name: "b", T: ddl.Type{Name: "STRING", Len: 0, IsArray: false}, NotNull: true, Comment: "", Id: "c8"},
+				},
+				Pks: []ddl.IndexKey{{Col: "a", Desc: false, Order: 1}},
+				Id:  "t6",
+			}},
+	}
+	assert.Equal(t, expectedConv.SpSchema, res.SpSchema)
+
+}
+
 func buildConvMySQL(conv *internal.Conv) {
 	conv.SrcSchema = map[string]schema.Table{
 		"t1": {