EXPERIMENTAL breaking down the vreplication flow

go/vt/sqlparser/comments.go

@@ -17,6 +17,7 @@ limitations under the License.
 package sqlparser
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 	"unicode"
@@ -277,6 +278,17 @@ func (d CommentDirectives) IsSet(key string) bool {
 	return false
 }
 
+// IsTrue checks the directive map for the named directive and returns
+// true if the directive is set and has a true value, which can be:
+// - true boolean
+// - 1 int
+// - a string that evaluates as true
+func (d CommentDirectives) IsTrue(key string) bool {
+	value := fmt.Sprintf("%v", d[key])
+	isTrue, _ := strconv.ParseBool(value)
+	return isTrue
+}
+
 // SkipQueryPlanCacheDirective returns true if skip query plan cache directive is set to true in query.
 func SkipQueryPlanCacheDirective(stmt Statement) bool {
 	switch stmt := stmt.(type) {

go/vt/sqlparser/comments_test.go

@@ -356,7 +356,9 @@ func TestExtractCommentDirectives(t *testing.T) {
 			}
 		})
 	}
+}
 
+func TestIsSet(t *testing.T) {
 	d := CommentDirectives{
 		"ONE_OPT": true,
 		"TWO_OPT": false,
@@ -369,28 +371,64 @@ func TestExtractCommentDirectives(t *testing.T) {
 	if !d.IsSet("ONE_OPT") {
 		t.Errorf("d.IsSet(ONE_OPT) should be true")
 	}
-
 	if d.IsSet("TWO_OPT") {
 		t.Errorf("d.IsSet(TWO_OPT) should be false")
 	}
-
 	if !d.IsSet("three") {
 		t.Errorf("d.IsSet(three) should be true")
 	}
-
 	if d.IsSet("four") {
 		t.Errorf("d.IsSet(four) should be false")
 	}
-
 	if d.IsSet("five") {
 		t.Errorf("d.IsSet(five) should be false")
 	}
-
 	if d.IsSet("six") {
 		t.Errorf("d.IsSet(six) should be false")
 	}
 }
 
+func TestIsTrue(t *testing.T) {
+	d := CommentDirectives{
+		"ONE_OPT": true,
+		"TWO_OPT": false,
+		"three":   1,
+		"four":    2,
+		"five":    0,
+		"six":     "true",
+		"seven":   "1",
+		"eight":   "0",
+	}
+
+	if !d.IsTrue("ONE_OPT") {
+		t.Errorf("d.IsSet(ONE_OPT) should be true")
+	}
+	if d.IsTrue("TWO_OPT") {
+		t.Errorf("d.IsSet(TWO_OPT) should be false")
+	}
+	if !d.IsTrue("three") {
+		t.Errorf("d.IsSet(three) should be true")
+	}
+	if d.IsTrue("four") {
+		t.Errorf("d.IsSet(four) should be false")
+	}
+	if d.IsTrue("five") {
+		t.Errorf("d.IsSet(five) should be false")
+	}
+	if !d.IsTrue("six") {
+		t.Errorf("d.IsSet(six) should be true")
+	}
+	if !d.IsTrue("seven") {
+		t.Errorf("d.IsSet(seven) should be true")
+	}
+	if d.IsTrue("eight") {
+		t.Errorf("d.IsSet(eight) should be false")
+	}
+	if d.IsTrue("no-such-key") {
+		t.Errorf("d.IsSet(no-such-key) should be false")
+	}
+}
+
 func TestSkipQueryPlanCacheDirective(t *testing.T) {
 	stmt, _ := Parse("insert /*vt+ SKIP_QUERY_PLAN_CACHE=1 */ into user(id) values (1), (2)")
 	if !SkipQueryPlanCacheDirective(stmt) {

go/vt/vttablet/tabletmanager/vreplication/external_connector_test.go

@@ -63,7 +63,7 @@ func TestExternalConnectorCopy(t *testing.T) {
 
 	expectDBClientAndVreplicationQueries(t, []string{
 		"begin",
-		"insert into tab1(id,val) values (1,'a'), (2,'b')",
+		"* into tab1(id,val) values (1,'a'), (2,'b')",
 		"/update _vt.copy_state",
 		"commit",
 		"/delete from _vt.copy_state",
@@ -72,7 +72,7 @@ func TestExternalConnectorCopy(t *testing.T) {
 	execStatements(t, []string{"insert into tab1 values(3, 'c')"})
 	expectDBClientQueries(t, []string{
 		"begin",
-		"insert into tab1(id,val) values (3,'c')",
+		"* into tab1(id,val) values (3,'c')",
 		"/update _vt.vreplication set pos=",
 		"commit",
 	})
@@ -162,8 +162,8 @@ func TestExternalConnectorPlay(t *testing.T) {
 
 	expectDBClientAndVreplicationQueries(t, []string{
 		"begin",
-		"insert into tab1(id,val) values (1,'a')",
-		"insert into tab1(id,val) values (2,'b')",
+		"* into tab1(id,val) values (1,'a')",
+		"* into tab1(id,val) values (2,'b')",
 		"/update _vt.vreplication set pos=",
 		"commit",
 	}, pos)

go/vt/vttablet/tabletmanager/vreplication/framework_test.go

@@ -497,6 +497,9 @@ func expectDBClientQueries(t *testing.T, queries []string, skippableOnce ...stri
 			}
 			return result
 		}
+		if query[0] == '*' {
+			return strings.Contains(got, query[1:])
+		}
 		return (got == query)
 	}
 	for i, query := range queries {
@@ -582,6 +585,8 @@ func expectNontxQueries(t *testing.T, queries []string) {
 					panic(err)
 				}
 				match = result
+			} else if query[0] == '*' {
+				match = strings.Contains(got, query[1:])
 			} else {
 				match = (got == query)
 			}

go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go

@@ -277,8 +277,17 @@ func buildTablePlan(tableName string, rule *binlogdatapb.Rule, colInfoMap map[st
 			},
 		})
 	}
-	sendRule.Filter = sqlparser.String(tpb.sendSelect)
 
+	strictSnapshot := false
+	switch tpb.onInsert {
+	case insertOnDup, insertIgnore:
+		strictSnapshot = true
+	}
+	comments := sqlparser.Comments{
+		fmt.Sprintf(`/*vt+ strictSnapshot=%v */`, strictSnapshot),
+	}
+	tpb.sendSelect.Comments = comments
+	sendRule.Filter = sqlparser.String(tpb.sendSelect)
 	tablePlan := tpb.generate()
 	tablePlan.SendRule = sendRule
 	tablePlan.EnumValuesMap = enumValuesMap
@@ -568,6 +577,9 @@ func (tpb *tablePlanBuilder) generateInsertStatement() *sqlparser.ParsedQuery {
 func (tpb *tablePlanBuilder) generateInsertPart(buf *sqlparser.TrackedBuffer) *sqlparser.ParsedQuery {
 	if tpb.onInsert == insertIgnore {
 		buf.Myprintf("insert ignore into %v(", tpb.name)
+	} else if tpb.onInsert == insertNormal {
+		// the condition (tpb.onInsert == insertNormal) is true when there is no GROUP BY
+		buf.Myprintf("replace into %v(", tpb.name)
 	} else {
 		buf.Myprintf("insert into %v(", tpb.name)
 	}

go/vt/vttablet/tabletmanager/vreplication/vcopier.go

@@ -108,7 +108,7 @@ func (vc *vcopier) initTablesForCopy(ctx context.Context) error {
 // copyNext also builds the copyState metadata that contains the tables and their last
 // primary key that was copied. A nil Result means that nothing has been copied.
 // A table that was fully copied is removed from copyState.
-func (vc *vcopier) copyNext(ctx context.Context, settings binlogplayer.VRSettings) error {
+func (vc *vcopier) copyNext(ctx context.Context) error {
 	qr, err := vc.vr.dbClient.Execute(fmt.Sprintf("select table_name, lastpk from _vt.copy_state where vrepl_id=%d", vc.vr.id))
 	if err != nil {
 		return err
@@ -324,6 +324,7 @@ func (vc *vcopier) copyTable(ctx context.Context, tableName string, copyState ma
 	if err != nil {
 		return err
 	}
+	// Getting here means we are certain we are done with the copy, and we delete copy_state
 	log.Infof("Copy of %v finished at lastpk: %v", tableName, bv)
 	buf := sqlparser.NewTrackedBuffer(nil)
 	buf.Myprintf("delete from _vt.copy_state where vrepl_id=%s and table_name=%s", strconv.Itoa(int(vc.vr.id)), encodeString(tableName))

go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go

@@ -117,10 +117,10 @@ func TestPlayerCopyCharPK(t *testing.T) {
 		"/update _vt.vreplication set message='Picked source tablet.*",
 		"/insert into _vt.copy_state",
 		"/update _vt.vreplication set state='Copying'",
-		"insert into dst(idc,val) values ('a\\0',1)",
+		"* into dst(idc,val) values ('a\\0',1)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"idc\\" type:BINARY} rows:{lengths:2 values:\\"a\\\\x00\\"}' where vrepl_id=.*`,
 		`update dst set val=3 where idc='a\0' and ('a\0') <= ('a\0')`,
-		"insert into dst(idc,val) values ('c\\0',2)",
+		"* into dst(idc,val) values ('c\\0',2)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"idc\\" type:BINARY} rows:{lengths:2 values:\\"c\\\\x00\\"}' where vrepl_id=.*`,
 		"/delete from _vt.copy_state.*dst",
 		"/update _vt.vreplication set state='Running'",
@@ -218,12 +218,12 @@ func TestPlayerCopyVarcharPKCaseInsensitive(t *testing.T) {
 		"/update _vt.vreplication set message='Picked source tablet.*",
 		"/insert into _vt.copy_state",
 		"/update _vt.vreplication set state='Copying'",
-		"insert into dst(idc,val) values ('a',1)",
+		"* into dst(idc,val) values ('a',1)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"idc\\" type:VARCHAR} rows:{lengths:1 values:\\"a\\"}' where vrepl_id=.*`,
-		`/insert into dst\(idc,val\) select 'B', 3 from dual where \( .* 'B' COLLATE .* \) <= \( .* 'a' COLLATE .* \)`,
-		"insert into dst(idc,val) values ('B',3)",
+		`/replace into dst\(idc,val\) select 'B', 3 from dual where \( .* 'B' COLLATE .* \) <= \( .* 'a' COLLATE .* \)`,
+		"* into dst(idc,val) values ('B',3)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"idc\\" type:VARCHAR} rows:{lengths:1 values:\\"B\\"}' where vrepl_id=.*`,
-		"insert into dst(idc,val) values ('c',2)",
+		"* into dst(idc,val) values ('c',2)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"idc\\" type:VARCHAR} rows:{lengths:1 values:\\"c\\"}' where vrepl_id=.*`,
 		"/delete from _vt.copy_state.*dst",
 		"/update _vt.vreplication set state='Running'",
@@ -322,10 +322,10 @@ func TestPlayerCopyVarcharCompositePKCaseSensitiveCollation(t *testing.T) {
 		"/update _vt.vreplication set message='Picked source tablet.*",
 		"/insert into _vt.copy_state",
 		"/update _vt.vreplication set state='Copying'",
-		"insert into dst(id,idc,idc2,val) values (1,'a','a',1)",
+		"* into dst(id,idc,idc2,val) values (1,'a','a',1)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} fields:{name:\\"idc\\" type:VARBINARY} fields:{name:\\"idc2\\" type:VARBINARY} rows:{lengths:1 lengths:1 lengths:1 values:\\"1aa\\"}' where vrepl_id=.*`,
-		`insert into dst(id,idc,idc2,val) select 1, 'B', 'B', 3 from dual where (1,'B','B') <= (1,'a','a')`,
-		"insert into dst(id,idc,idc2,val) values (1,'c','c',2)",
+		`replace into dst(id,idc,idc2,val) select 1, 'B', 'B', 3 from dual where (1,'B','B') <= (1,'a','a')`,
+		"* into dst(id,idc,idc2,val) values (1,'c','c',2)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} fields:{name:\\"idc\\" type:VARBINARY} fields:{name:\\"idc2\\" type:VARBINARY} rows:{lengths:1 lengths:1 lengths:1 values:\\"1cc\\"}' where vrepl_id=.*`,
 		"/delete from _vt.copy_state.*dst",
 		"/update _vt.vreplication set state='Running'",
@@ -395,7 +395,7 @@ func TestPlayerCopyTablesWithFK(t *testing.T) {
 		// The first fast-forward has no starting point. So, it just saves the current position.
 		"/update _vt.vreplication set pos=",
 		"begin",
-		"insert into dst1(id,id2) values (1,1), (2,2)",
+		"* into dst1(id,id2) values (1,1), (2,2)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		"commit",
 		// copy of dst1 is done: delete from copy_state.
@@ -407,7 +407,7 @@ func TestPlayerCopyTablesWithFK(t *testing.T) {
 		"commit",
 		// copy dst2
 		"begin",
-		"insert into dst2(id,id2) values (1,21), (2,22)",
+		"* into dst2(id,id2) values (1,21), (2,22)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		"commit",
 		// copy of dst1 is done: delete from copy_state.
@@ -500,7 +500,7 @@ func TestPlayerCopyTables(t *testing.T) {
 		// The first fast-forward has no starting point. So, it just saves the current position.
 		"/update _vt.vreplication set pos=",
 		"begin",
-		"insert into dst1(id,val) values (1,'aaa'), (2,'bbb')",
+		"* into dst1(id,val) values (1,'aaa'), (2,'bbb')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		"commit",
 		// copy of dst1 is done: delete from copy_state.
@@ -631,16 +631,16 @@ func TestPlayerCopyBigTable(t *testing.T) {
 		"/insert into _vt.copy_state",
 		// The first fast-forward has no starting point. So, it just saves the current position.
 		"/update _vt.vreplication set state='Copying'",
-		"insert into dst(id,val) values (1,'aaa')",
+		"* into dst(id,val) values (1,'aaa')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"1\\"}' where vrepl_id=.*`,
 		// The next catchup executes the new row insert, but will be a no-op.
-		"insert into dst(id,val) select 3, 'ccc' from dual where (3) <= (1)",
+		"* into dst(id,val) select 3, 'ccc' from dual where (3) <= (1)",
 		// fastForward has nothing to add. Just saves position.
 		// Second row gets copied.
-		"insert into dst(id,val) values (2,'bbb')",
+		"* into dst(id,val) values (2,'bbb')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		// Third row copied without going back to catchup state.
-		"insert into dst(id,val) values (3,'ccc')",
+		"* into dst(id,val) values (3,'ccc')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"3\\"}' where vrepl_id=.*`,
 		"/delete from _vt.copy_state.*dst",
 		// Copy is done. Go into running state.
@@ -746,16 +746,16 @@ func TestPlayerCopyWildcardRule(t *testing.T) {
 		"/insert into _vt.copy_state",
 		"/update _vt.vreplication set state='Copying'",
 		// The first fast-forward has no starting point. So, it just saves the current position.
-		"insert into src(id,val) values (1,'aaa')",
+		"* into src(id,val) values (1,'aaa')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"1\\"}' where vrepl_id=.*`,
 		// The next catchup executes the new row insert, but will be a no-op.
-		"insert into src(id,val) select 3, 'ccc' from dual where (3) <= (1)",
+		"* into src(id,val) select 3, 'ccc' from dual where (3) <= (1)",
 		// fastForward has nothing to add. Just saves position.
 		// Second row gets copied.
-		"insert into src(id,val) values (2,'bbb')",
+		"* into src(id,val) values (2,'bbb')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		// Third row copied without going back to catchup state.
-		"insert into src(id,val) values (3,'ccc')",
+		"* into src(id,val) values (3,'ccc')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"3\\"}' where vrepl_id=.*`,
 		"/delete from _vt.copy_state.*src",
 		// Copy is done. Go into running state.
@@ -883,23 +883,23 @@ func TestPlayerCopyTableContinuation(t *testing.T) {
 	expectNontxQueries(t, []string{
 		// Catchup
 		"/update _vt.vreplication set message='Picked source tablet.*",
-		"insert into dst1(id,val) select 1, 'insert in' from dual where (1,1) <= (6,6)",
-		"insert into dst1(id,val) select 7, 'insert out' from dual where (7,7) <= (6,6)",
+		"* into dst1(id,val) select 1, 'insert in' from dual where (1,1) <= (6,6)",
+		"* into dst1(id,val) select 7, 'insert out' from dual where (7,7) <= (6,6)",
 		"update dst1 set val='updated' where id=3 and (3,3) <= (6,6)",
 		"update dst1 set val='updated' where id=10 and (10,10) <= (6,6)",
 		"delete from dst1 where id=4 and (4,4) <= (6,6)",
 		"delete from dst1 where id=9 and (9,9) <= (6,6)",
 		"delete from dst1 where id=5 and (5,5) <= (6,6)",
-		"insert into dst1(id,val) select 5, 'move within' from dual where (5,10) <= (6,6)",
+		"* into dst1(id,val) select 5, 'move within' from dual where (5,10) <= (6,6)",
 		"delete from dst1 where id=6 and (6,6) <= (6,6)",
-		"insert into dst1(id,val) select 12, 'move out' from dual where (12,6) <= (6,6)",
+		"* into dst1(id,val) select 12, 'move out' from dual where (12,6) <= (6,6)",
 		"delete from dst1 where id=11 and (11,11) <= (6,6)",
-		"insert into dst1(id,val) select 4, 'move in' from dual where (4,11) <= (6,6)",
+		"* into dst1(id,val) select 4, 'move in' from dual where (4,11) <= (6,6)",
 		"update copied set val='bbb' where id=1",
 		// Fast-forward
 		"update dst1 set val='updated again' where id=3 and (3,3) <= (6,6)",
 		// Copy
-		"insert into dst1(id,val) values (7,'insert out'), (8,'no change'), (10,'updated'), (12,'move out')",
+		"* into dst1(id,val) values (7,'insert out'), (8,'no change'), (10,'updated'), (12,'move out')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id1\\" type:INT32} fields:{name:\\"id2\\" type:INT32} rows:{lengths:2 lengths:1 values:\\"126\\"}' where vrepl_id=.*`,
 		"/delete from _vt.copy_state.*dst1",
 		// Copy again. There should be no events for catchup.
@@ -995,9 +995,9 @@ func TestPlayerCopyWildcardTableContinuation(t *testing.T) {
 		"/insert into _vt.vreplication",
 		"/update _vt.vreplication set state = 'Copying'",
 		"/update _vt.vreplication set message='Picked source tablet.*",
-		"insert into dst(id,val) select 4, 'new' from dual where (4) <= (2)",
+		"* into dst(id,val) select 4, 'new' from dual where (4) <= (2)",
 		// Copy
-		"insert into dst(id,val) values (3,'uncopied'), (4,'new')",
+		"* into dst(id,val) values (3,'uncopied'), (4,'new')",
 		`/update _vt.copy_state set lastpk.*`,
 		"/delete from _vt.copy_state.*dst",
 		"/update _vt.vreplication set state='Running'",
@@ -1082,7 +1082,7 @@ func TestPlayerCopyWildcardTableContinuationWithOptimizeInserts(t *testing.T) {
 		"/update _vt.vreplication set state = 'Copying'",
 		"/update _vt.vreplication set message='Picked source tablet.*",
 		// Copy
-		"insert into dst(id,val) values (3,'uncopied'), (4,'new')",
+		"* into dst(id,val) values (3,'uncopied'), (4,'new')",
 		`/update _vt.copy_state set lastpk.*`,
 		"/delete from _vt.copy_state.*dst",
 		"/update _vt.vreplication set state='Running'",
@@ -1188,7 +1188,7 @@ func TestPlayerCopyTablesStopAfterCopy(t *testing.T) {
 		// The first fast-forward has no starting point. So, it just saves the current position.
 		"/update _vt.vreplication set pos=",
 		"begin",
-		"insert into dst1(id,val) values (1,'aaa'), (2,'bbb')",
+		"* into dst1(id,val) values (1,'aaa'), (2,'bbb')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		"commit",
 		// copy of dst1 is done: delete from copy_state.
@@ -1267,7 +1267,7 @@ func TestPlayerCopyTableCancel(t *testing.T) {
 		// The first fast-forward has no starting point. So, it just saves the current position.
 		"/update _vt.vreplication set pos=",
 		"begin",
-		"insert into dst1(id,val) values (1,'aaa'), (2,'bbb')",
+		"* into dst1(id,val) values (1,'aaa'), (2,'bbb')",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		"commit",
 		// copy of dst1 is done: delete from copy_state.