Merge pull request #180 from Shopify/allow-existing-table

Allow target table to already exist in CopydbFerry

copydb/config.go

@@ -70,6 +70,11 @@ type Config struct {
 
 	// The duration to wait for the replication to catchup before aborting. Only use if RunFerryFromReplica is true.
 	WaitForReplicationTimeout string
+
+	// Ghostferry will by default create tables on your target and fail if these already exist.
+	// This allows pre-existing tables on your target, schemas and data compatibility won't be validated.
+	// Use at your own risk.
+	AllowExistingTargetTable bool
 }
 
 func (c *Config) InitializeAndValidateConfig() error {

copydb/copydb.go

@@ -159,7 +159,7 @@ func (this *CopydbFerry) createDatabaseIfExistsOnTarget(database string) error {
 }
 
 func (this *CopydbFerry) createTableOnTarget(database, table string) error {
-	var tableNameAgain, createTableQuery string
+	var tableNameAgain, createTableQuery, createStatement string
 
 	r := this.Ferry.SourceDB.QueryRow(fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`", database, table))
 	err := r.Scan(&tableNameAgain, &createTableQuery)
@@ -180,10 +180,16 @@ func (this *CopydbFerry) createTableOnTarget(database, table string) error {
 		tableNameAgain = targetTableName
 	}
 
+	if this.config.AllowExistingTargetTable {
+		createStatement = "CREATE TABLE IF NOT EXISTS `%s`.`%s`"
+	} else {
+		createStatement = "CREATE TABLE `%s`.`%s`"
+	}
+
 	createTableQueryReplaced := strings.Replace(
 		createTableQuery,
 		fmt.Sprintf("CREATE TABLE `%s`", table),
-		fmt.Sprintf("CREATE TABLE `%s`.`%s`", database, tableNameAgain),
+		fmt.Sprintf(createStatement, database, tableNameAgain),
 		1,
 	)
 

copydb/test/copydb_test.go

@@ -85,6 +85,29 @@ func (t *CopydbTestSuite) TestCreateDatabaseAndTableWithRewrites() {
 	t.Require().Equal(renamedTableName, value)
 }
 
+func (t *CopydbTestSuite) TestCreateDatabasesAndTablesAlreadyExists() {
+	var err error
+	t.copydbFerry.Ferry.Tables, err = ghostferry.LoadTables(t.ferry.SourceDB, t.copydbFerry.Ferry.TableFilter, nil, nil, nil)
+	t.Require().Nil(err)
+
+	testhelpers.SeedInitialData(t.ferry.TargetDB, renamedSchemaName, renamedTableName, 1)
+
+	err = t.copydbFerry.CreateDatabasesAndTables()
+	t.Require().EqualError(err, "Error 1050: Table 'test_table_1_renamed' already exists")
+}
+
+func (t *CopydbTestSuite) TestCreateDatabasesAndTablesAlreadyExistsAllowed() {
+	var err error
+	t.copydbFerry.Ferry.Tables, err = ghostferry.LoadTables(t.ferry.SourceDB, t.copydbFerry.Ferry.TableFilter, nil, nil, nil)
+	t.Require().Nil(err)
+
+	testhelpers.SeedInitialData(t.ferry.TargetDB, renamedSchemaName, renamedTableName, 1)
+	t.copydbConfig.AllowExistingTargetTable = true
+
+	err = t.copydbFerry.CreateDatabasesAndTables()
+	t.Require().Nil(err)
+}
+
 func (t *CopydbTestSuite) TestCreateDatabaseAndTableWithOrdering() {
 	// NOTE: Here we just ensure passing a table does not cause issues in the
 	// invocation. A more thorough test is done in the table-schema tests