Repository: vitessio/vitess
Pull request #9988 (2022-03-27 06:52:27)

More testing for vreplication unique key violation protection

## Description

This PR adds a endtoend test for the `vreplication` suite to validate a unique key violation scenario. The expecteation is for `vreplication` to fail the migration due to a unique key.

## Related Issue(s)

- #6926 

## Checklist
- [ ] Should this PR be backported?
- [x] Tests were added or are not required
- [ ] Documentation was added or is not required

## Deployment Notes
<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->


Patch:
diff --git a/go/vt/vtgate/planbuilder/physical/subquery_planning.go b/go/vt/vtgate/planbuilder/physical/subquery_planning.go
index 2f31f40013..4abcb3c434 100644
--- a/go/vt/vtgate/planbuilder/physical/subquery_planning.go
+++ b/go/vt/vtgate/planbuilder/physical/subquery_planning.go
@@ -4,6 +4,7 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/engine"
+	"vitess.io/vitess/go/vt/vtgate/evalengine"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/abstract"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 
@@ -100,6 +101,9 @@ func mergeSubQueryOp(ctx *plancontext.PlanningContext, outer *Route, inner *Rout
 	}
 	outer.SysTableTableSchema = append(outer.SysTableTableSchema, inner.SysTableTableSchema...)
 	for k, v := range inner.SysTableTableName {
+		if outer.SysTableTableName == nil {
+			outer.SysTableTableName = map[string]evalengine.Expr{}
+		}
 		outer.SysTableTableName[k] = v
 	}
 
diff --git a/go/vt/vtgate/planbuilder/route.go b/go/vt/vtgate/planbuilder/route.go
index 2d74934b06..e54da457ba 100644
--- a/go/vt/vtgate/planbuilder/route.go
+++ b/go/vt/vtgate/planbuilder/route.go
@@ -380,6 +380,9 @@ func (rb *route) MergeSubquery(pb *primitiveBuilder, inner *route) bool {
 			case engine.DBA, engine.Reference:
 				rb.eroute.SysTableTableSchema = append(rb.eroute.SysTableTableSchema, inner.eroute.SysTableTableSchema...)
 				for k, v := range inner.eroute.SysTableTableName {
+					if rb.eroute.SysTableTableName == nil {
+						rb.eroute.SysTableTableName = map[string]evalengine.Expr{}
+					}
 					rb.eroute.SysTableTableName[k] = v
 				}
 				rb.eroute.Opcode = engine.DBA
diff --git a/go/vt/wrangler/materializer.go b/go/vt/wrangler/materializer.go
index 394bcb8d92..755a9ef04b 100644
--- a/go/vt/wrangler/materializer.go
+++ b/go/vt/wrangler/materializer.go
@@ -30,6 +30,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"vitess.io/vitess/go/json2"
+	"vitess.io/vitess/go/sqlescape"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/binlog/binlogplayer"
 	"vitess.io/vitess/go/vt/concurrency"
@@ -454,11 +455,11 @@ func (wr *Wrangler) prepareCreateLookup(ctx context.Context, keyspace string, sp
 	if !strings.Contains(vindex.Type, "lookup") {
 		return nil, nil, nil, fmt.Errorf("vindex %s is not a lookup type", vindex.Type)
 	}
-	strs := strings.Split(vindex.Params["table"], ".")
-	if len(strs) != 2 {
-		return nil, nil, nil, fmt.Errorf("vindex 'table' must be <keyspace>.<table>: %v", vindex)
+
+	targetKeyspace, targetTableName, err = sqlparser.ParseTable(vindex.Params["table"])
+	if err != nil || targetKeyspace == "" {
+		return nil, nil, nil, fmt.Errorf("vindex table name must be in the form <keyspace>.<table>. Got: %v", vindex.Params["table"])
 	}
-	targetKeyspace, targetTableName = strs[0], strs[1]
 
 	vindexFromCols = strings.Split(vindex.Params["from"], ",")
 	if strings.Contains(vindex.Type, "unique") {
@@ -593,15 +594,15 @@ func (wr *Wrangler) prepareCreateLookup(ctx context.Context, keyspace string, sp
 	}
 
 	if vindex.Params["data_type"] == "" || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") {
-		modified = append(modified, fmt.Sprintf("  `%s` varbinary(128),", vindexToCol))
+		modified = append(modified, fmt.Sprintf("  %s varbinary(128),", sqlescape.EscapeID(vindexToCol)))
 	} else {
-		modified = append(modified, fmt.Sprintf("  `%s` `%s`,", vindexToCol, vindex.Params["data_type"]))
+		modified = append(modified, fmt.Sprintf("  %s %s,", sqlescape.EscapeID(vindexToCol), sqlescape.EscapeID(vindex.Params["data_type"])))
 	}
 	buf := sqlparser.NewTrackedBuffer(nil)
 	fmt.Fprintf(buf, "  PRIMARY KEY (")
 	prefix := ""
 	for _, col := range vindexFromCols {
-		fmt.Fprintf(buf, "%s`%s`", prefix, col)
+		fmt.Fprintf(buf, "%s%s", prefix, sqlescape.EscapeID(col))
 		prefix = ", "
 	}
 	fmt.Fprintf(buf, ")")
@@ -699,8 +700,9 @@ func (wr *Wrangler) prepareCreateLookup(ctx context.Context, keyspace string, sp
 }
 
 func generateColDef(lines []string, sourceVindexCol, vindexFromCol string) (string, error) {
-	source := fmt.Sprintf("`%s`", sourceVindexCol)
-	target := fmt.Sprintf("`%s`", vindexFromCol)
+	source := sqlescape.EscapeID(sourceVindexCol)
+	target := sqlescape.EscapeID(vindexFromCol)
+
 	for _, line := range lines[1:] {
 		if strings.Contains(line, source) {
 			line = strings.Replace(line, source, target, 1)
@@ -727,12 +729,11 @@ func (wr *Wrangler) ExternalizeVindex(ctx context.Context, qualifiedVindexName s
 	if sourceVindex == nil {
 		return fmt.Errorf("vindex %s not found in vschema", qualifiedVindexName)
 	}
-	qualifiedTableName := sourceVindex.Params["table"]
-	splits = strings.Split(qualifiedTableName, ".")
-	if len(splits) != 2 {
-		return fmt.Errorf("table name in vindex should be of the form keyspace.table: %s", qualifiedTableName)
+
+	targetKeyspace, targetTableName, err := sqlparser.ParseTable(sourceVindex.Params["table"])
+	if err != nil || targetKeyspace == "" {
+		return fmt.Errorf("vindex table name must be in the form <keyspace>.<table>. Got: %v", sourceVindex.Params["table"])
 	}
-	targetKeyspace, targetTableName := splits[0], splits[1]
 	workflow := targetTableName + "_vdx"
 	targetShards, err := wr.ts.GetServingShards(ctx, targetKeyspace)
 	if err != nil {


Test patch:
diff --git a/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go b/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go
index 7e8d105f76..ae814600a4 100644
--- a/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go
+++ b/go/test/endtoend/onlineddl/vrepl_stress/onlineddl_vrepl_mini_stress_test.go
@@ -82,24 +82,20 @@ deletesAttempts=%d, deletesFailures=%d, deletesNoops=%d, deletes=%d,
 }
 
 var (
-	clusterInstance      *cluster.LocalProcessCluster
-	shards               []cluster.Shard
-	vtParams             mysql.ConnParams
-	evaluatedMysqlParams *mysql.ConnParams
+	clusterInstance *cluster.LocalProcessCluster
+	shards          []cluster.Shard
+	vtParams        mysql.ConnParams
 
 	opOrder               int64
 	opOrderMutex          sync.Mutex
-	onlineDDLStrategy     = "online -vreplication-test-suite"
+	onlineDDLStrategy     = "vitess"
 	hostname              = "localhost"
 	keyspaceName          = "ks"
 	cell                  = "zone1"
 	schemaChangeDirectory = ""
 	tableName             = `stress_test`
-	afterTableName        = `stress_test_after`
 	cleanupStatements     = []string{
 		`DROP TABLE IF EXISTS stress_test`,
-		`DROP TABLE IF EXISTS stress_test_before`,
-		`DROP TABLE IF EXISTS stress_test_after`,
 	}
 	createStatement = `
 		CREATE TABLE stress_test (
@@ -126,32 +122,13 @@ var (
 	deleteRowStatement = `
 		DELETE FROM stress_test WHERE id=%d AND updates=1
 	`
+	selectMaxOpOrder = `
+		SELECT MAX(op_order) as m FROM stress_test
+	`
 	// We use CAST(SUM(updates) AS SIGNED) because SUM() returns a DECIMAL datatype, and we want to read a SIGNED INTEGER type
 	selectCountRowsStatement = `
 		SELECT COUNT(*) AS num_rows, CAST(SUM(updates) AS SIGNED) AS sum_updates FROM stress_test
 	`
-	// We use CAST(SUM(updates) AS SIGNED) because SUM() returns a DECIMAL datatype, and we want to read a SIGNED INTEGER type
-	selectCountRowsFromAfterTableStatement = `
-		SELECT COUNT(*) AS num_rows, CAST(SUM(updates) AS SIGNED) AS sum_updates FROM stress_test_after
-	`
-	selectCountFromTableBefore = `
-		SELECT count(*) as c FROM stress_test_before
-	`
-	selectCountFromTableAfter = `
-		SELECT count(*) as c FROM stress_test_after
-	`
-	selectMaxOpOrderFromTableBefore = `
-		SELECT MAX(op_order) as m FROM stress_test_before
-	`
-	selectMaxOpOrderFromTableAfter = `
-		SELECT MAX(op_order) as m FROM stress_test_after
-	`
-	selectBeforeTable = `
-		SELECT * FROM stress_test_before order by id
-	`
-	selectAfterTable = `
-		SELECT * FROM stress_test_after order by id
-	`
 	truncateStatement = `
 		TRUNCATE TABLE stress_test
 	`
@@ -178,22 +155,6 @@ func nextOpOrder() int64 {
 	return opOrder
 }
 
-func getTablet() *cluster.Vttablet {
-	return clusterInstance.Keyspaces[0].Shards[0].Vttablets[0]
-}
-
-func mysqlParams() *mysql.ConnParams {
-	if evaluatedMysqlParams != nil {
-		return evaluatedMysqlParams
-	}
-	evaluatedMysqlParams = &mysql.ConnParams{
-		Uname:      "vt_dba",
-		UnixSocket: path.Join(os.Getenv("VTDATAROOT"), fmt.Sprintf("/vt_%010d", getTablet().TabletUID), "/mysql.sock"),
-		DbName:     fmt.Sprintf("vt_%s", keyspaceName),
-	}
-	return evaluatedMysqlParams
-}
-
 func TestMain(m *testing.M) {
 	defer cluster.PanicHandler(nil)
 	flag.Parse()
@@ -311,7 +272,7 @@ func TestSchemaChange(t *testing.T) {
 		hint := "hint-alter-without-workload"
 		uuid := testOnlineDDLStatement(t, fmt.Sprintf(alterHintStatement, hint), onlineDDLStrategy, "vtgate", hint)
 		onlineddl.CheckMigrationStatus(t, &vtParams, shards, uuid, schema.OnlineDDLStatusComplete)
-		testSelectTableMetricsAfterMigration(t)
+		testSelectTableMetrics(t)
 	})
 
 	for i := 0; i < countIterations; i++ {
@@ -344,7 +305,7 @@ func TestSchemaChange(t *testing.T) {
 				wg.Wait()
 			})
 			t.Run("validate metrics", func(t *testing.T) {
-				testSelectTableMetricsAfterMigration(t)
+				testSelectTableMetrics(t)
 			})
 		})
 	}
@@ -388,7 +349,7 @@ func testOnlineDDLStatement(t *testing.T, alterStatement string, ddlStrategy str
 	}
 
 	if expectHint != "" {
-		checkMigratedTable(t, afterTableName, expectHint)
+		checkMigratedTable(t, tableName, expectHint)
 	}
 	return uuid
 }
@@ -523,12 +484,6 @@ func runSingleConnection(ctx context.Context, t *testing.T, done *int64) {
 		case 2:
 			err = generateDelete(t, conn)
 		}
-		if err != nil {
-			if strings.Contains(err.Error(), "doesn't exist") {
-				// Table renamed to _before, due to -vreplication-test-suite flag
-				err = nil
-			}
-		}
 		assert.Nil(t, err)
 		time.Sleep(singleConnectionSleepInterval)
 	}
@@ -585,7 +540,19 @@ func initTable(t *testing.T) {
 	}
 }
 
-func testSelectTableMetricsWithStatement(t *testing.T, statement string) {
+func testSelectTableMetrics(t *testing.T) {
+	writeMetrics.mu.Lock()
+	defer writeMetrics.mu.Unlock()
+
+	{
+		rs := onlineddl.VtgateExecQuery(t, &vtParams, selectMaxOpOrder, "")
+		row := rs.Named().Row()
+		require.NotNil(t, row)
+
+		maxOpOrder := row.AsInt64("m", 0)
+		fmt.Printf("# max op_order in table: %d\n", maxOpOrder)
+	}
+
 	log.Infof("%s", writeMetrics.String())
 
 	ctx := context.Background()
@@ -593,7 +560,7 @@ func testSelectTableMetricsWithStatement(t *testing.T, statement string) {
 	require.Nil(t, err)
 	defer conn.Close()
 
-	rs, err := conn.ExecuteFetch(statement, 1000, true)
+	rs, err := conn.ExecuteFetch(selectCountRowsStatement, 1000, true)
 	require.Nil(t, err)
 
 	row := rs.Named().Row()
@@ -601,83 +568,11 @@ func testSelectTableMetricsWithStatement(t *testing.T, statement string) {
 	log.Infof("testSelectTableMetrics, row: %v", row)
 	numRows := row.AsInt64("num_rows", 0)
 	sumUpdates := row.AsInt64("sum_updates", 0)
-
 	assert.NotZero(t, numRows)
 	assert.NotZero(t, sumUpdates)
 	assert.NotZero(t, writeMetrics.inserts)
 	assert.NotZero(t, writeMetrics.deletes)
 	assert.NotZero(t, writeMetrics.updates)
 	assert.Equal(t, writeMetrics.inserts-writeMetrics.deletes, numRows)
-}
-
-func testSelectTableMetrics(t *testing.T) {
-	testSelectTableMetricsWithStatement(t, selectCountRowsStatement)
-}
-
-func testSelectTableMetricsAfterMigration(t *testing.T) {
-	writeMetrics.mu.Lock()
-	defer writeMetrics.mu.Unlock()
-
-	var countBefore int64
-	{
-		// Validate after table is populated
-		rs := onlineddl.VtgateExecQuery(t, &vtParams, selectCountFromTableBefore, "")
-		row := rs.Named().Row()
-		require.NotNil(t, row)
-
-		countBefore = row.AsInt64("c", 0)
-		require.NotZero(t, countBefore)
-		require.Less(t, countBefore, int64(maxTableRows))
-
-		fmt.Printf("# count rows in table (before): %d\n", countBefore)
-	}
-	var countAfter int64
-	{
-		// Validate after table is populated
-		rs := onlineddl.VtgateExecQuery(t, &vtParams, selectCountFromTableAfter, "")
-		row := rs.Named().Row()
-		require.NotNil(t, row)
-
-		countAfter = row.AsInt64("c", 0)
-		require.NotZero(t, countAfter)
-		require.Less(t, countAfter, int64(maxTableRows))
-
-		fmt.Printf("# count rows in table (after): %d\n", countAfter)
-	}
-	{
-		rs := onlineddl.VtgateExecQuery(t, &vtParams, selectMaxOpOrderFromTableBefore, "")
-		row := rs.Named().Row()
-		require.NotNil(t, row)
-
-		maxOpOrder := row.AsInt64("m", 0)
-		fmt.Printf("# max op_order in table (before): %d\n", maxOpOrder)
-	}
-	{
-		rs := onlineddl.VtgateExecQuery(t, &vtParams, selectMaxOpOrderFromTableAfter, "")
-		row := rs.Named().Row()
-		require.NotNil(t, row)
-
-		maxOpOrder := row.AsInt64("m", 0)
-		fmt.Printf("# max op_order in table (after): %d\n", maxOpOrder)
-	}
-
-	testSelectTableMetricsWithStatement(t, selectCountRowsFromAfterTableStatement)
-
-	{
-		selectBeforeFile := onlineddl.CreateTempScript(t, selectBeforeTable)
-		defer os.Remove(selectBeforeFile)
-		beforeOutput := onlineddl.MysqlClientExecFile(t, mysqlParams(), os.TempDir(), "", selectBeforeFile)
-		beforeOutput = strings.TrimSpace(beforeOutput)
-		require.NotEmpty(t, beforeOutput)
-		assert.Equal(t, countBefore, int64(len(strings.Split(beforeOutput, "\n"))))
-
-		selectAfterFile := onlineddl.CreateTempScript(t, selectAfterTable)
-		defer os.Remove(selectAfterFile)
-		afterOutput := onlineddl.MysqlClientExecFile(t, mysqlParams(), os.TempDir(), "", selectAfterFile)
-		afterOutput = strings.TrimSpace(afterOutput)
-		require.NotEmpty(t, afterOutput)
-		assert.Equal(t, countAfter, int64(len(strings.Split(afterOutput, "\n"))))
-
-		require.Equal(t, beforeOutput, afterOutput, "results mismatch: (%s) and (%s)", selectBeforeTable, selectAfterTable)
-	}
+	assert.Equal(t, writeMetrics.updates-writeMetrics.deletes, sumUpdates) // because we DELETE WHERE updates=1
 }
diff --git a/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go b/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go
index 139ec61b29..0f8ea5a7a6 100644
--- a/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go
+++ b/go/test/endtoend/onlineddl/vrepl_stress_suite/onlineddl_vrepl_stress_suite_test.go
@@ -64,6 +64,8 @@ type testcase struct {
 	expectAddedUniqueKeys int64
 	// expectRemovedUniqueKeys is the number of alleviated constraints
 	expectRemovedUniqueKeys int64
+	// autoIncInsert is a special case where we don't generate id values. It's a specific test case.
+	autoIncInsert bool
 }
 
 var (
@@ -108,6 +110,12 @@ var (
 			prepareStatement: "",
 			alterStatement:   "engine=innodb",
 		},
+		{
+			name:             "autoinc PK",
+			prepareStatement: "modify id bigint not null auto_increment",
+			alterStatement:   "engine=innodb",
+			autoIncInsert:    true,
+		},
 		{
 			name:             "UK similar to PK, no PK",
 			prepareStatement: "add unique key id_uidx(id)",
@@ -321,6 +329,9 @@ var (
 		alter table stress_test modify hint_col varchar(64) not null default '%s'
 	`
 
+	insertRowAutoIncStatement = `
+		INSERT IGNORE INTO stress_test (id, id_negative, rand_text, rand_num, op_order) VALUES (NULL, %d, concat(left(md5(%d), 8), '_', %d), floor(rand()*1000000), %d)
+	`
 	insertRowStatement = `
 		INSERT IGNORE INTO stress_test (id, id_negative, rand_text, rand_num, op_order) VALUES (%d, %d, concat(left(md5(%d), 8), '_', %d), floor(rand()*1000000), %d)
 	`
@@ -506,7 +517,7 @@ func TestSchemaChange(t *testing.T) {
 				wg.Add(1)
 				go func() {
 					defer wg.Done()
-					runMultipleConnections(ctx, t)
+					runMultipleConnections(ctx, t, testcase.autoIncInsert)
 				}()
 				uuid := testOnlineDDLStatement(t, fullStatement, onlineDDLStrategy, "vtgate", hintText)
 				expectStatus := schema.OnlineDDLStatusComplete
@@ -517,7 +528,7 @@ func TestSchemaChange(t *testing.T) {
 				cancel() // will cause runMultipleConnections() to terminate
 				wg.Wait()
 				if !testcase.expectFailure {
-					testCompareBeforeAfterTables(t)
+					testCompareBeforeAfterTables(t, testcase.autoIncInsert)
 				}
 
 				rs := onlineddl.ReadMigrations(t, &vtParams, uuid)
@@ -608,9 +619,13 @@ func getCreateTableStatement(t *testing.T, tablet *cluster.Vttablet, tableName s
 	return statement
 }
 
-func generateInsert(t *testing.T, conn *mysql.Conn) error {
+func generateInsert(t *testing.T, conn *mysql.Conn, autoIncInsert bool) error {
 	id := rand.Int31n(int32(maxTableRows))
 	query := fmt.Sprintf(insertRowStatement, id, -id, id, id, nextOpOrder())
+	if autoIncInsert {
+		id = rand.Int31()
+		query = fmt.Sprintf(insertRowAutoIncStatement, -id, id, id, nextOpOrder())
+	}
 	qr, err := conn.ExecuteFetch(query, 1000, true)
 	if err == nil && qr != nil {
 		assert.Less(t, qr.RowsAffected, uint64(2))
@@ -638,7 +653,7 @@ func generateDelete(t *testing.T, conn *mysql.Conn) error {
 	return err
 }
 
-func runSingleConnection(ctx context.Context, t *testing.T, done *int64) {
+func runSingleConnection(ctx context.Context, t *testing.T, autoIncInsert bool, done *int64) {
 	log.Infof("Running single connection")
 	conn, err := mysql.Connect(ctx, &vtParams)
 	require.Nil(t, err)
@@ -656,7 +671,7 @@ func runSingleConnection(ctx context.Context, t *testing.T, done *int64) {
 		}
 		switch rand.Int31n(3) {
 		case 0:
-			err = generateInsert(t, conn)
+			err = generateInsert(t, conn, autoIncInsert)
 		case 1:
 			err = generateUpdate(t, conn)
 		case 2:
@@ -673,7 +688,7 @@ func runSingleConnection(ctx context.Context, t *testing.T, done *int64) {
 	}
 }
 
-func runMultipleConnections(ctx context.Context, t *testing.T) {
+func runMultipleConnections(ctx context.Context, t *testing.T, autoIncInsert bool) {
 	log.Infof("Running multiple connections")
 	var done int64
 	var wg sync.WaitGroup
@@ -681,7 +696,7 @@ func runMultipleConnections(ctx context.Context, t *testing.T) {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			runSingleConnection(ctx, t, &done)
+			runSingleConnection(ctx, t, autoIncInsert, &done)
 		}()
 	}
 	<-ctx.Done()
@@ -706,7 +721,7 @@ func initTable(t *testing.T) {
 	require.Nil(t, err)
 
 	for i := 0; i < maxTableRows/2; i++ {
-		generateInsert(t, conn)
+		generateInsert(t, conn, false)
 	}
 	for i := 0; i < maxTableRows/4; i++ {
 		generateUpdate(t, conn)
@@ -730,7 +745,7 @@ func initTable(t *testing.T) {
 }
 
 // testCompareBeforeAfterTables validates that stress_test_before and stress_test_after contents are non empty and completely identical
-func testCompareBeforeAfterTables(t *testing.T) {
+func testCompareBeforeAfterTables(t *testing.T, autoIncInsert bool) {
 	var countBefore int64
 	{
 		// Validate after table is populated
@@ -740,8 +755,9 @@ func testCompareBeforeAfterTables(t *testing.T) {
 
 		countBefore = row.AsInt64("c", 0)
 		require.NotZero(t, countBefore)
-		require.Less(t, countBefore, int64(maxTableRows))
-
+		if !autoIncInsert {
+			require.Less(t, countBefore, int64(maxTableRows))
+		}
 		fmt.Printf("# count rows in table (before): %d\n", countBefore)
 	}
 	var countAfter int64
@@ -753,8 +769,9 @@ func testCompareBeforeAfterTables(t *testing.T) {
 
 		countAfter = row.AsInt64("c", 0)
 		require.NotZero(t, countAfter)
-		require.Less(t, countAfter, int64(maxTableRows))
-
+		if !autoIncInsert {
+			require.Less(t, countAfter, int64(maxTableRows))
+		}
 		fmt.Printf("# count rows in table (after): %d\n", countAfter)
 	}
 	{
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/alter b/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/alter
new file mode 100644
index 0000000000..a6e49dea04
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/alter
@@ -0,0 +1,1 @@
+add column empty_text varchar(5) not null default '', add unique key uk_empty_text(empty_text)
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/create.sql b/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/create.sql
new file mode 100644
index 0000000000..3a7d2ad744
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/create.sql
@@ -0,0 +1,10 @@
+drop table if exists onlineddl_test;
+create table onlineddl_test (
+  id int auto_increment,
+  val int not null,
+  ts timestamp,
+  primary key(id)
+) auto_increment=1;
+
+insert into onlineddl_test values (null, 2, now());
+insert into onlineddl_test values (null, 3, now());
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/expect_failure b/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/expect_failure
new file mode 100644
index 0000000000..3a1e3ef6af
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/fail-duplicate-entry2/expect_failure
@@ -0,0 +1,1 @@
+Duplicate entry
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index c7d52f6e80..1de98a26d5 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -2965,3 +2965,40 @@ Gen4 plan same as above
   }
 }
 Gen4 plan same as above
+
+# dual query with exists clause
+"select 1 from dual where exists (select 1 from information_schema.TABLES where information_schema.TABLES.TABLE_NAME = 'proc' and information_schema.TABLES.TABLE_SCHEMA = 'mysql')"
+{
+  "QueryType": "SELECT",
+  "Original": "select 1 from dual where exists (select 1 from information_schema.TABLES where information_schema.TABLES.TABLE_NAME = 'proc' and information_schema.TABLES.TABLE_SCHEMA = 'mysql')",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "DBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select 1 from dual where 1 != 1",
+    "Query": "select 1 from dual where exists (select 1 from information_schema.`TABLES` where information_schema.`TABLES`.TABLE_NAME = :TABLES_TABLE_NAME and information_schema.`TABLES`.TABLE_SCHEMA = :__vtschemaname)",
+    "SysTableTableName": "[TABLES_TABLE_NAME:VARCHAR(\"proc\")]",
+    "SysTableTableSchema": "[VARCHAR(\"mysql\")]",
+    "Table": "dual"
+  }
+}
+{
+  "QueryType": "SELECT",
+  "Original": "select 1 from dual where exists (select 1 from information_schema.TABLES where information_schema.TABLES.TABLE_NAME = 'proc' and information_schema.TABLES.TABLE_SCHEMA = 'mysql')",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "DBA",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "FieldQuery": "select 1 from dual where 1 != 1",
+    "Query": "select 1 from dual where exists (select 1 from information_schema.`TABLES` where `TABLES`.TABLE_NAME = :TABLES_TABLE_NAME and `TABLES`.TABLE_SCHEMA = :__vtschemaname)",
+    "SysTableTableName": "[TABLES_TABLE_NAME:VARCHAR(\"proc\")]",
+    "SysTableTableSchema": "[VARCHAR(\"mysql\")]",
+    "Table": "dual"
+  }
+}
diff --git a/go/vt/wrangler/materializer_test.go b/go/vt/wrangler/materializer_test.go
index d132e28e07..9e58e5e1bd 100644
--- a/go/vt/wrangler/materializer_test.go
+++ b/go/vt/wrangler/materializer_test.go
@@ -1436,7 +1436,7 @@ func TestCreateLookupVindexFailures(t *testing.T) {
 				},
 			},
 		},
-		err: "vindex 'table' must be <keyspace>.<table>",
+		err: "vindex table name must be in the form <keyspace>.<table>",
 	}, {
 		description: "unique lookup should have only one from column",
 		input: &vschemapb.Keyspace{
@@ -1698,7 +1698,7 @@ func TestExternalizeVindex(t *testing.T) {
 		err:   "vindex sourceks.absent not found in vschema",
 	}, {
 		input: "sourceks.bad",
-		err:   "table name in vindex should be of the form keyspace.table: unqualified",
+		err:   "vindex table name must be in the form <keyspace>.<table>. Got: unqualified",
 	}, {
 		input:        "sourceks.owned",
 		vrResponse:   running,