VReplication: Defer Secondary Index Creation

go/mysql/constants.go

@@ -424,6 +424,7 @@ const (
 	ErSPNotVarArg                   = 1414
 	ERRowIsReferenced2              = 1451
 	ErNoReferencedRow2              = 1452
+	ERDupIndex                      = 1831
 	ERInnodbReadOnly                = 1874
 
 	// already exists

go/test/endtoend/vreplication/config_test.go

@@ -32,22 +32,23 @@ package vreplication
 // The internal table _vt_PURGE_4f9194b43b2011eb8a0104ed332e05c2_20221210194431 should be ignored by vreplication
 // The db_order_test table is used to ensure vreplication and vdiff work well with complex non-integer PKs, even across DB versions.
 var (
+	// All standard user tables should have a primary key and at least one secondary key.
 	initialProductSchema = `
-create table product(pid int, description varbinary(128), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key(pid)) CHARSET=utf8mb4;
+create table product(pid int, description varbinary(128), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key(pid), key(date1,date2)) CHARSET=utf8mb4;
 create table customer(cid int, name varchar(128) collate utf8mb4_bin, meta json default null, typ enum('individual','soho','enterprise'), sport set('football','cricket','baseball'),
 	ts timestamp not null default current_timestamp, bits bit(2) default b'11', date1 datetime not null default '0000-00-00 00:00:00', 
-	date2 datetime not null default '2021-00-01 00:00:00', dec80 decimal(8,0), primary key(cid,typ)) CHARSET=utf8mb4;
+	date2 datetime not null default '2021-00-01 00:00:00', dec80 decimal(8,0), primary key(cid,typ), key(name)) CHARSET=utf8mb4;
 create table customer_seq(id int, next_id bigint, cache bigint, primary key(id)) comment 'vitess_sequence';
-create table merchant(mname varchar(128), category varchar(128), primary key(mname)) CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
-create table orders(oid int, cid int, pid int, mname varchar(128), price int, qty int, total int as (qty * price), total2 int as (qty * price) stored, primary key(oid)) CHARSET=utf8;
+create table merchant(mname varchar(128), category varchar(128), primary key(mname), key(category)) CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
+create table orders(oid int, cid int, pid int, mname varchar(128), price int, qty int, total int as (qty * price), total2 int as (qty * price) stored, primary key(oid), key(pid), key(cid)) CHARSET=utf8;
 create table order_seq(id int, next_id bigint, cache bigint, primary key(id)) comment 'vitess_sequence';
-create table customer2(cid int, name varchar(128), typ enum('individual','soho','enterprise'), sport set('football','cricket','baseball'),ts timestamp not null default current_timestamp, primary key(cid)) CHARSET=utf8;
+create table customer2(cid int, name varchar(128), typ enum('individual','soho','enterprise'), sport set('football','cricket','baseball'),ts timestamp not null default current_timestamp, primary key(cid), key(ts)) CHARSET=utf8;
 create table customer_seq2(id int, next_id bigint, cache bigint, primary key(id)) comment 'vitess_sequence';
-create table ` + "`Lead`(`Lead-id`" + ` binary(16), name varbinary(16), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key (` + "`Lead-id`" + `));
-create table ` + "`Lead-1`(`Lead`" + ` binary(16), name varbinary(16), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key (` + "`Lead`" + `));
-create table _vt_PURGE_4f9194b43b2011eb8a0104ed332e05c2_20221210194431(id int, val varbinary(128), primary key(id));
-create table db_order_test (c_uuid varchar(64) not null default '', created_at datetime not null, dstuff varchar(128), dtstuff text, dbstuff blob, cstuff char(32), primary key (c_uuid,created_at)) CHARSET=utf8mb4;
-create table datze (id int, dt1 datetime not null default current_timestamp, dt2 datetime not null, ts1 timestamp default current_timestamp, primary key (id));
+create table ` + "`Lead`(`Lead-id`" + ` binary(16), name varbinary(16), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key (` + "`Lead-id`" + `), key (date1));
+create table ` + "`Lead-1`(`Lead`" + ` binary(16), name varbinary(16), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key (` + "`Lead`" + `), key (date2));
+create table _vt_PURGE_4f9194b43b2011eb8a0104ed332e05c2_20221210194431(id int, val varbinary(128), primary key(id), key(val));
+create table db_order_test (c_uuid varchar(64) not null default '', created_at datetime not null, dstuff varchar(128), dtstuff text, dbstuff blob, cstuff char(32), primary key (c_uuid,created_at), key (dstuff)) CHARSET=utf8mb4;
+create table datze (id int, dt1 datetime not null default current_timestamp, dt2 datetime not null, ts1 timestamp default current_timestamp, primary key (id), key (dt1));
 `
 
 	// These should always be ignored in vreplication

go/test/endtoend/vreplication/helper_test.go

@@ -29,20 +29,19 @@ import (
 	"testing"
 	"time"
 
-	"vitess.io/vitess/go/vt/log"
-
+	"github.com/PuerkitoBio/goquery"
 	"github.com/buger/jsonparser"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/tidwall/gjson"
 
-	"vitess.io/vitess/go/test/endtoend/cluster"
-	"vitess.io/vitess/go/vt/schema"
-
-	"github.com/PuerkitoBio/goquery"
-
 	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/sqlescape"
 	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/test/endtoend/cluster"
+	"vitess.io/vitess/go/vt/log"
+	"vitess.io/vitess/go/vt/schema"
+	"vitess.io/vitess/go/vt/sqlparser"
 )
 
 const (
@@ -298,6 +297,57 @@ func waitForWorkflowState(t *testing.T, vc *VitessCluster, ksWorkflow string, wa
 	}
 }
 
+// confirmTablesHaveSecondaryKeys confirms that the tables provided
+// as a CSV have secondary keys. This is useful when testing the
+// --defer-secondary-keys flag to confirm that the secondary keys
+// were re-added by the time the workflow hits the running phase.
+// For a Reshard workflow, where no tables are specififed, pass
+// an empty string for the tables and all tables in the target
+// keyspace will be checked.
+func confirmTablesHaveSecondaryKeys(t *testing.T, tablets []*cluster.VttabletProcess, ksName string, tables string) {
+	require.NotNil(t, tablets)
+	require.NotNil(t, tablets[0])
+	var tableArr []string
+	if strings.TrimSpace(tables) != "" {
+		tableArr = strings.Split(tables, ",")
+	}
+	if len(tableArr) == 0 { // We don't specify any for Reshard.
+		// In this case we check all of them.
+		res, err := tablets[0].QueryTablet("show tables", ksName, true)
+		require.NoError(t, err)
+		require.NotNil(t, res)
+		for _, row := range res.Rows {
+			tableArr = append(tableArr, row[0].ToString())
+		}
+	}
+	for _, tablet := range tablets {
+		for _, table := range tableArr {
+			if schema.IsInternalOperationTableName(table) {
+				continue
+			}
+			table := strings.TrimSpace(table)
+			secondaryKeys := 0
+			res, err := tablet.QueryTablet(fmt.Sprintf("show create table %s", sqlescape.EscapeID(table)), ksName, true)
+			require.NoError(t, err)
+			require.NotNil(t, res)
+			row := res.Named().Row()
+			tableSchema := row["Create Table"].ToString()
+			parsedDDL, err := sqlparser.ParseStrictDDL(tableSchema)
+			require.NoError(t, err)
+			createTable, ok := parsedDDL.(*sqlparser.CreateTable)
+			require.True(t, ok)
+			require.NotNil(t, createTable)
+			require.NotNil(t, createTable.GetTableSpec())
+			for _, index := range createTable.GetTableSpec().Indexes {
+				if !index.Info.Primary {
+					secondaryKeys++
+				}
+			}
+			require.Greater(t, secondaryKeys, 0, "Table %s does not have any secondary keys", table)
+		}
+	}
+}
+
 func getHTTPBody(url string) string {
 	resp, err := http.Get(url)
 	if err != nil {

go/test/endtoend/vreplication/resharding_workflows_v2_test.go

@@ -63,6 +63,7 @@ func createReshardWorkflow(t *testing.T, sourceShards, targetShards string) erro
 		"", workflowActionCreate, "", sourceShards, targetShards)
 	require.NoError(t, err)
 	waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
+	confirmTablesHaveSecondaryKeys(t, []*cluster.VttabletProcess{targetTab1}, targetKs, "")
 	catchup(t, targetTab1, workflowName, "Reshard")
 	catchup(t, targetTab2, workflowName, "Reshard")
 	vdiff1(t, ksWorkflow, "")
@@ -77,6 +78,7 @@ func createMoveTablesWorkflow(t *testing.T, tables string) {
 		tables, workflowActionCreate, "", "", "")
 	require.NoError(t, err)
 	waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
+	confirmTablesHaveSecondaryKeys(t, []*cluster.VttabletProcess{targetTab1}, targetKs, tables)
 	catchup(t, targetTab1, workflowName, "MoveTables")
 	catchup(t, targetTab2, workflowName, "MoveTables")
 	vdiff1(t, ksWorkflow, "")
@@ -110,6 +112,11 @@ func tstWorkflowExec(t *testing.T, cells, workflow, sourceKs, targetKs, tables,
 		} else {
 			args = append(args, "--source_shards", sourceShards, "--target_shards", targetShards)
 		}
+		// Test new experimental --defer-secondary-keys flag
+		switch currentWorkflowType {
+		case wrangler.MoveTablesWorkflow, wrangler.MigrateWorkflow, wrangler.ReshardWorkflow:
+			args = append(args, "--defer-secondary-keys")
+		}
 	}
 	if cells != "" {
 		args = append(args, "--cells", cells)

go/test/endtoend/vreplication/vreplication_test.go

@@ -220,7 +220,6 @@ func TestVreplicationCopyThrottling(t *testing.T) {
 		// to avoid flakiness when the CI is very slow.
 		fmt.Sprintf("--queryserver-config-transaction-timeout=%d", int64(defaultTimeout.Seconds())*3),
 		fmt.Sprintf("--vreplication_copy_phase_max_innodb_history_list_length=%d", maxSourceTrxHistory),
-
 		parallelInsertWorkers,
 	}
 

go/vt/binlog/binlogplayer/binlog_player.go

@@ -577,6 +577,7 @@ var AlterVReplicationTable = []string{
 	"ALTER TABLE _vt.vreplication ADD COLUMN time_throttled BIGINT NOT NULL DEFAULT 0",
 	"ALTER TABLE _vt.vreplication ADD COLUMN component_throttled VARCHAR(255) NOT NULL DEFAULT ''",
 	"ALTER TABLE _vt.vreplication ADD COLUMN workflow_sub_type int NOT NULL DEFAULT 0",
+	"ALTER TABLE _vt.vreplication ADD COLUMN defer_secondary_keys bool NOT NULL DEFAULT false",
 }
 
 // WithDDLInitialQueries contains the queries that:
@@ -596,20 +597,21 @@ var WithDDLInitialQueries = []string{
 
 // VRSettings contains the settings of a vreplication table.
 type VRSettings struct {
-	StartPos          mysql.Position
-	StopPos           mysql.Position
-	MaxTPS            int64
-	MaxReplicationLag int64
-	State             string
-	WorkflowType      int32
-	WorkflowSubType   int32
-	WorkflowName      string
+	StartPos           mysql.Position
+	StopPos            mysql.Position
+	MaxTPS             int64
+	MaxReplicationLag  int64
+	State              string
+	WorkflowType       int32
+	WorkflowSubType    int32
+	WorkflowName       string
+	DeferSecondaryKeys bool
 }
 
 // ReadVRSettings retrieves the throttler settings for
 // vreplication from the checkpoint table.
 func ReadVRSettings(dbClient DBClient, uid uint32) (VRSettings, error) {
-	query := fmt.Sprintf("select pos, stop_pos, max_tps, max_replication_lag, state, workflow_type, workflow, workflow_sub_type from _vt.vreplication where id=%v", uid)
+	query := fmt.Sprintf("select pos, stop_pos, max_tps, max_replication_lag, state, workflow_type, workflow, workflow_sub_type, defer_secondary_keys from _vt.vreplication where id=%v", uid)
 	qr, err := dbClient.ExecuteFetch(query, 1)
 	if err != nil {
 		return VRSettings{}, fmt.Errorf("error %v in selecting vreplication settings %v", err, query)
@@ -646,27 +648,32 @@ func ReadVRSettings(dbClient DBClient, uid uint32) (VRSettings, error) {
 	if err != nil {
 		return VRSettings{}, fmt.Errorf("failed to parse workflow_sub_type column: %v", err)
 	}
+	deferSecondaryKeys, err := vrRow.ToBool("defer_secondary_keys")
+	if err != nil {
+		return VRSettings{}, fmt.Errorf("failed to parse defer_secondary_keys column: %v", err)
+	}
 	return VRSettings{
-		StartPos:          startPos,
-		StopPos:           stopPos,
-		MaxTPS:            maxTPS,
-		MaxReplicationLag: maxReplicationLag,
-		State:             vrRow.AsString("state", ""),
-		WorkflowType:      workflowType,
-		WorkflowName:      vrRow.AsString("workflow", ""),
-		WorkflowSubType:   workflowSubType,
+		StartPos:           startPos,
+		StopPos:            stopPos,
+		MaxTPS:             maxTPS,
+		MaxReplicationLag:  maxReplicationLag,
+		State:              vrRow.AsString("state", ""),
+		WorkflowType:       workflowType,
+		WorkflowName:       vrRow.AsString("workflow", ""),
+		WorkflowSubType:    workflowSubType,
+		DeferSecondaryKeys: deferSecondaryKeys,
 	}, nil
 }
 
 // CreateVReplication returns a statement to populate the first value into
 // the _vt.vreplication table.
 func CreateVReplication(workflow string, source *binlogdatapb.BinlogSource, position string, maxTPS, maxReplicationLag, timeUpdated int64, dbName string,
-	workflowType binlogdatapb.VReplicationWorkflowType, workflowSubType binlogdatapb.VReplicationWorkflowSubType) string {
+	workflowType binlogdatapb.VReplicationWorkflowType, workflowSubType binlogdatapb.VReplicationWorkflowSubType, deferSecondaryKeys bool) string {
 	return fmt.Sprintf("insert into _vt.vreplication "+
-		"(workflow, source, pos, max_tps, max_replication_lag, time_updated, transaction_timestamp, state, db_name, workflow_type, workflow_sub_type) "+
-		"values (%v, %v, %v, %v, %v, %v, 0, '%v', %v, %v, %v)",
+		"(workflow, source, pos, max_tps, max_replication_lag, time_updated, transaction_timestamp, state, db_name, workflow_type, workflow_sub_type, defer_secondary_keys) "+
+		"values (%v, %v, %v, %v, %v, %v, 0, '%v', %v, %v, %v, %v)",
 		encodeString(workflow), encodeString(source.String()), encodeString(position), maxTPS, maxReplicationLag,
-		timeUpdated, BlpRunning, encodeString(dbName), int64(workflowType), int64(workflowSubType))
+		timeUpdated, BlpRunning, encodeString(dbName), int64(workflowType), int64(workflowSubType), deferSecondaryKeys)
 }
 
 // CreateVReplicationState returns a statement to create a stopped vreplication.