Use table info

pkg/restore/db.go

@@ -45,7 +45,27 @@ func NewDB(g glue.Glue, store kv.Storage) (*DB, error) {
 // ExecDDL executes the query of a ddl job.
 func (db *DB) ExecDDL(ctx context.Context, ddlJob *model.Job) error {
 	var err error
-	if ddlJob.BinlogInfo.TableInfo != nil {
+	tableInfo := ddlJob.BinlogInfo.TableInfo
+	dbInfo := ddlJob.BinlogInfo.DBInfo
+	switch ddlJob.Type {
+	case model.ActionCreateSchema:
+		err = db.se.CreateDatabase(ctx, dbInfo)
+		if err != nil {
+			log.Error("create database failed", zap.Stringer("db", dbInfo.Name), zap.Error(err))
+		}
+		return errors.Trace(err)
+	case model.ActionCreateTable:
+		err = db.se.CreateTable(ctx, model.NewCIStr(ddlJob.SchemaName), tableInfo)
+		if err != nil {
+			log.Error("create table failed",
+				zap.Stringer("db", dbInfo.Name),
+				zap.Stringer("table", tableInfo.Name),
+				zap.Error(err))
+		}
+		return errors.Trace(err)
+	}
+
+	if tableInfo != nil {
 		switchDbSQL := fmt.Sprintf("use %s;", utils.EncloseName(ddlJob.SchemaName))
 		err = db.se.Execute(ctx, switchDbSQL)
 		if err != nil {

pkg/task/restore.go

@@ -471,6 +471,10 @@ func enableTiDBConfig() {
 	conf.MaxIndexLength = config.DefMaxOfMaxIndexLength
 	log.Warn("set max-index-length to max(3072*4) to skip check index length in DDL")
 
+	// we need set this to true, since all create table DDLs will create with tableInfo
+	// and we can handle alter drop pk/add pk DDLs with no impact
+	conf.AlterPrimaryKey = true
+
 	// set this to true for some auto random DDL execute normally during incremental restore
 	conf.Experimental.AllowAutoRandom = true
 	conf.Experimental.AllowsExpressionIndex = true

tests/br_incompatible_tidb_config/run.sh

@@ -29,17 +29,33 @@ run_sql "create schema $DB;"
 
 # test alter pk issue https://github.com/pingcap/br/issues/215
 TABLE="t1"
+INCREMENTAL_TABLE="t1inc"
 
 run_sql "create table $DB.$TABLE (a int primary key, b int unique);"
 run_sql "insert into $DB.$TABLE values (42, 42);"
 
 # backup
 run_br --pd $PD_ADDR backup db --db "$DB" -s "local://$TEST_DIR/$DB$TABLE"
 
+run_sql "create table $DB.$INCREMENTAL_TABLE (a int primary key, b int unique);"
+run_sql "insert into $DB.$INCREMENTAL_TABLE values (42, 42);"
+
+# drop pk
+run_sql "alter table $DB.$INCREMENTAL_TABLE drop primary key"
+run_sql "drop table $DB.$INCREMENTAL_TABLE"
+run_sql "create table $DB.$INCREMENTAL_TABLE like $DB.$TABLE"
+run_sql "insert into $DB.$INCREMENTAL_TABLE values (42, 42);"
+
+# incremental backup
+run_br --pd $PD_ADDR backup db --db "$DB" -s "local://$TEST_DIR/$DB$INCREMENTAL_TABLE"
+
 # restore
 run_sql "drop schema $DB;"
+
 run_br --pd $PD_ADDR restore db --db "$DB" -s "local://$TEST_DIR/$DB$TABLE"
 
+run_br --pd $PD_ADDR restore db --db "$DB" -s "local://$TEST_DIR/$DB$INCREMENTAL_TABLE"
+
 run_sql "drop schema $DB;"
 run_sql "create schema $DB;"