restore: fix alter auto increment id for no-primary-key table (#139) (#…

…142)

* restore: fix alter auto increment id for no-primary-key table

Signed-off-by: 5kbpers <[email protected]>

* add integration test

Signed-off-by: 5kbpers <[email protected]>

* address comments

Signed-off-by: 5kbpers <[email protected]>

Co-authored-by: 5kbpers <[email protected]>

pkg/restore/db.go

@@ -92,6 +92,19 @@ func (db *DB) CreateTable(ctx context.Context, table *utils.Table) error {
 			zap.Error(err))
 		return errors.Trace(err)
 	}
+	alterAutoIncIDSQL := fmt.Sprintf(
+		"alter table %s auto_increment = %d",
+		escapeTableName(schema.Name),
+		schema.AutoIncID)
+	_, err = db.se.Execute(ctx, alterAutoIncIDSQL)
+	if err != nil {
+		log.Error("alter AutoIncID failed",
+			zap.String("SQL", alterAutoIncIDSQL),
+			zap.Stringer("db", table.Db.Name),
+			zap.Stringer("table", table.Schema.Name),
+			zap.Error(err))
+		return errors.Trace(err)
+	}
 	return nil
 }
 

pkg/restore/db_test.go

@@ -41,23 +41,23 @@ func (s *testRestoreSchemaSuite) TestRestoreAutoIncID(c *C) {
 	tk := testkit.NewTestKit(c, s.mock.Storage)
 	tk.MustExec("use test")
 	tk.MustExec("set @@sql_mode=''")
-	tk.MustExec("drop table if exists t;")
+	tk.MustExec("drop table if exists `\"t\"`;")
 	// Test SQL Mode
-	tk.MustExec("create table t (" +
-		"a int not null auto_increment," +
+	tk.MustExec("create table `\"t\"` (" +
+		"a int not null," +
 		"time timestamp not null default '0000-00-00 00:00:00'," +
 		"primary key (a));",
 	)
-	tk.MustExec("insert into t values (10, '0000-00-00 00:00:00');")
+	tk.MustExec("insert into `\"t\"` values (10, '0000-00-00 00:00:00');")
 	// Query the current AutoIncID
-	autoIncID, err := strconv.ParseUint(tk.MustQuery("admin show t next_row_id").Rows()[0][3].(string), 10, 64)
+	autoIncID, err := strconv.ParseUint(tk.MustQuery("admin show `\"t\"` next_row_id").Rows()[0][3].(string), 10, 64)
 	c.Assert(err, IsNil, Commentf("Error query auto inc id: %s", err))
 	// Get schemas of db and table
 	info, err := s.mock.Domain.GetSnapshotInfoSchema(math.MaxUint64)
 	c.Assert(err, IsNil, Commentf("Error get snapshot info schema: %s", err))
 	dbInfo, exists := info.SchemaByName(model.NewCIStr("test"))
 	c.Assert(exists, IsTrue, Commentf("Error get db info"))
-	tableInfo, err := info.TableByName(model.NewCIStr("test"), model.NewCIStr("t"))
+	tableInfo, err := info.TableByName(model.NewCIStr("test"), model.NewCIStr("\"t\""))
 	c.Assert(err, IsNil, Commentf("Error get table info: %s", err))
 	table := utils.Table{
 		Schema: tableInfo.Meta(),
@@ -88,7 +88,7 @@ func (s *testRestoreSchemaSuite) TestRestoreAutoIncID(c *C) {
 	c.Assert(err, IsNil, Commentf("Error create table: %s %s", err, s.mock.DSN))
 	tk.MustExec("use test")
 	// Check if AutoIncID is altered successfully
-	autoIncID, err = strconv.ParseUint(tk.MustQuery("admin show t next_row_id").Rows()[0][3].(string), 10, 64)
+	autoIncID, err = strconv.ParseUint(tk.MustQuery("admin show `\"t\"` next_row_id").Rows()[0][3].(string), 10, 64)
 	c.Assert(err, IsNil, Commentf("Error query auto inc id: %s", err))
 	c.Assert(autoIncID, Equals, uint64(globalAutoID+100))
 }

pkg/restore/util.go

@@ -368,3 +368,12 @@ func encodeKeyPrefix(key []byte) []byte {
 	encodedPrefix = append(encodedPrefix, codec.EncodeBytes([]byte{}, key[:len(key)-ungroupedLen])...)
 	return append(encodedPrefix[:len(encodedPrefix)-9], key[len(key)-ungroupedLen:]...)
 }
+
+// escape the identifier for pretty-printing.
+// For instance, the identifier "foo `bar`" will become "`foo ``bar```".
+// The sqlMode controls whether to escape with backquotes (`) or double quotes
+// (`"`) depending on whether mysql.ModeANSIQuotes is enabled.
+func escapeTableName(cis model.CIStr) string {
+	quote := "`"
+	return quote + strings.Replace(cis.O, quote, quote+quote, -1) + quote
+}

tests/br_insert_after_restore/run.sh

@@ -0,0 +1,82 @@
+#!/bin/sh
+#
+# Copyright 2019 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -eu
+DB="$TEST_NAME"
+TABLE="usertable"
+ROW_COUNT=10
+PATH="tests/$TEST_NAME:bin:$PATH"
+
+insertRecords() {
+    for i in $(seq $1); do
+        run_sql "INSERT INTO $DB.$TABLE VALUES ('$i');"
+    done
+}
+
+createTable() {
+    run_sql "CREATE TABLE IF NOT EXISTS $DB.$TABLE (c1 CHAR(255));"
+}
+
+echo "load data..."
+echo "create database"
+run_sql "CREATE DATABASE IF NOT EXISTS $DB;"
+echo "create table"
+createTable
+echo "insert records"
+insertRecords $ROW_COUNT
+
+row_count_ori=$(run_sql "SELECT COUNT(*) FROM $DB.$TABLE;" | awk '/COUNT/{print $2}')
+
+# backup full
+echo "backup start..."
+run_br --pd $PD_ADDR backup full -s "local://$TEST_DIR/$DB" --ratelimit 5 --concurrency 4
+
+run_sql "DROP DATABASE $DB;"
+
+# restore full
+echo "restore start..."
+run_br restore full -s "local://$TEST_DIR/$DB" --pd $PD_ADDR
+
+row_count_new=$(run_sql "SELECT COUNT(*) FROM $DB.$TABLE;" | awk '/COUNT/{print $2}')
+
+fail=false
+if [ "${row_count_ori}" != "${row_count_new}" ];then
+    fail=true
+    echo "TEST: [$TEST_NAME] fail on database $DB"
+fi
+echo "database $DB [original] row count: ${row_count_ori}, [after br] row count: ${row_count_new}"
+
+if $fail; then
+    echo "TEST: [$TEST_NAME] failed!"
+    exit 1
+fi
+
+# insert records
+insertRecords $ROW_COUNT
+row_count_insert=$(run_sql "SELECT COUNT(*) FROM $DB.$TABLE;" | awk '/COUNT/{print $2}')
+fail=false
+if [ "${row_count_insert}" != "$(expr $row_count_new \* 2)" ];then
+    fail=true
+    echo "TEST: [$TEST_NAME] fail on inserting records to database $DB after restore: ${row_count_insert}"
+fi
+
+if $fail; then
+    echo "TEST: [$TEST_NAME] failed!"
+    exit 1
+else
+    echo "TEST: [$TEST_NAME] successed!"
+fi
+
+run_sql "DROP DATABASE $DB;"