DO NOT MERGE: replica transaction

go/cmd/vtgate/status.go

@@ -38,13 +38,13 @@ func addStatusParts(vtg *vtgate.VTGate) {
 	servenv.AddStatusPart("Gateway Status", vtgate.StatusTemplate, func() interface{} {
 		return vtg.GetGatewayCacheStatus()
 	})
-	if *vtgate.GatewayImplementation == vtgate.GatewayImplementationDiscovery {
+	if vtgate.UsingLegacyGateway() {
 		servenv.AddStatusPart("Health Check Cache", discovery.LegacyHealthCheckTemplate, func() interface{} {
 			return legacyHealthCheck.CacheStatus()
 		})
 	} else {
 		servenv.AddStatusPart("Health Check Cache", discovery.HealthCheckTemplate, func() interface{} {
-			return vtg.Gateway().HealthCheck().CacheStatus()
+			return vtg.Gateway().TabletsCacheStatus()
 		})
 	}
 }

go/mysql/fakesqldb/server.go

@@ -531,6 +531,11 @@ func (db *DB) QueryLog() string {
 	return strings.Join(db.querylog, ";")
 }
 
+//ResetQueryLog resets the query log
+func (db *DB) ResetQueryLog() {
+	db.querylog = nil
+}
+
 // EnableConnFail makes connection to this fake DB fail.
 func (db *DB) EnableConnFail() {
 	db.mu.Lock()

go/test/endtoend/clustertest/vtgate_test.go

@@ -43,7 +43,7 @@ func TestVtgateProcess(t *testing.T) {
 	defer conn.Close()
 
 	exec(t, conn, "insert into customer(id, email) values(1,'email1')")
-
+	_ = exec(t, conn, "begin")
 	qr := exec(t, conn, "select id, email from customer")
 	if got, want := fmt.Sprintf("%v", qr.Rows), `[[INT64(1) VARCHAR("email1")]]`; got != want {
 		t.Errorf("select:\n%v want\n%v", got, want)

go/test/endtoend/tabletgateway/healthcheck/main_test.go

@@ -84,9 +84,10 @@ func TestMain(m *testing.M) {
 			return 1
 		}
 
-		// Start vtgate
 		vtgateInstance := clusterInstance.GetVtgateInstance()
+		// set the gateway we want to use
 		vtgateInstance.GatewayImplementation = "tabletgateway"
+		// Start vtgate
 		err = vtgateInstance.Setup()
 		if err != nil {
 			return 1

go/test/endtoend/tabletgateway/healthcheck/vtgate_test.go

@@ -36,7 +36,7 @@ import (
 	"vitess.io/vitess/go/test/endtoend/cluster"
 )
 
-func TestVtgateProcess(t *testing.T) {
+func TestVtgateHealthCheck(t *testing.T) {
 	defer cluster.PanicHandler(t)
 	// Healthcheck interval on tablet is set to 1s, so sleep for 2s
 	time.Sleep(2 * time.Second)
@@ -46,12 +46,8 @@ func TestVtgateProcess(t *testing.T) {
 	require.Nil(t, err)
 	defer conn.Close()
 
-	exec(t, conn, "insert into customer(id, email) values(1,'email1')")
-
-	qr := exec(t, conn, "select id, email from customer")
-	assert.Equal(t, fmt.Sprintf("%v", qr.Rows), `[[INT64(1) VARCHAR("email1")]]`, "select returned wrong result")
-	qr = exec(t, conn, "show vitess_tablets")
-	assert.Equal(t, len(qr.Rows), 3, "wrong number of results from show")
+	qr := exec(t, conn, "show vitess_tablets", "")
+	assert.Equal(t, 3, len(qr.Rows), "wrong number of results from show")
 }
 
 func verifyVtgateVariables(t *testing.T, url string) {
@@ -78,6 +74,89 @@ func verifyVtgateVariables(t *testing.T, url string) {
 	assert.True(t, isMasterTabletPresent(healthCheckConnection), "Atleast one master tablet needs to be present")
 }
 
+/*
+-begin on replica should explicitly say read only
+-tabletserver planner should stop dml (if easy and reasonable)
+-vtgate planbuilder should not send dml to replicas
+*/
+
+func TestReplicaTransactions(t *testing.T) {
+	// TODO(deepthi): this test seems to depend on previous test. Fix tearDown so that tests are independent
+	defer cluster.PanicHandler(t)
+	// Healthcheck interval on tablet is set to 1s, so sleep for 2s
+	time.Sleep(2 * time.Second)
+	ctx := context.Background()
+	masterConn, err := mysql.Connect(ctx, &vtParams)
+	require.NoError(t, err)
+	replicaConn, err := mysql.Connect(ctx, &vtParams)
+	require.NoError(t, err)
+	defer masterConn.Close()
+	defer replicaConn.Close()
+
+	// insert a row using master
+	exec(t, masterConn, "insert into customer(id, email) values(1,'email1')", "")
+	time.Sleep(1 * time.Second) // we sleep for a bit to make sure that the replication catches up
+
+	// after a short pause, SELECT the data inside a tx on a replica
+	exec(t, replicaConn, "use @replica", "")
+	// begin transaction on replica
+	exec(t, replicaConn, "begin", "")
+	qr := exec(t, replicaConn, "select id, email from customer", "")
+	assert.Equal(t, `[[INT64(1) VARCHAR("email1")]]`, fmt.Sprintf("%v", qr.Rows), "select returned wrong result")
+
+	// insert more data on master using a transaction
+	exec(t, masterConn, "begin", "")
+	exec(t, masterConn, "insert into customer(id, email) values(2,'email2')", "")
+	exec(t, masterConn, "commit", "")
+	time.Sleep(1 * time.Second)
+
+	// replica doesn't see new row because it is in a transaction
+	qr2 := exec(t, replicaConn, "select id, email from customer", "")
+	assert.Equal(t, qr.Rows, qr2.Rows)
+
+	// replica should see new row after closing the transaction
+	exec(t, replicaConn, "commit", "")
+
+	qr3 := exec(t, replicaConn, "select id, email from customer", "")
+	assert.Equal(t, `[[INT64(1) VARCHAR("email1")] [INT64(2) VARCHAR("email2")]]`, fmt.Sprintf("%v", qr3.Rows), "we are not seeing the updates after closing the replica transaction")
+	// since we can't do INSERT/DELETE/UPDATE, commit and rollback both just close the transaction
+	exec(t, replicaConn, "rollback", "")
+
+	// bring down the tablet and try to select again
+	// Error: ERROR 1105 (HY000): vtgate: http://localhost:15001/: vttablet: rpc error: code = Unavailable desc = all SubConns are in TransientFailure,
+	// latest connection error: connection error: desc = "transport: Error while dialing dial tcp 127.0.1.1:16101: connect: connection refused"
+
+	//mysql> select id from customer;
+	//ERROR 1105 (HY000): vtgate: http://deepthi-ThinkPad:15001/: vttablet: rpc error: code = Unavailable desc = all SubConns are in TransientFailure, latest connection error: connection error: desc = "transport: Error while dialing dial tcp 127.0.1.1:16101: connect: connection refused"
+	//mysql> commit;
+	//ERROR 1105 (HY000): vtgate: http://deepthi-ThinkPad:15001/: target: commerce.0.replica: vttablet: Connection Closed
+	//mysql> select id from customer;
+	//ERROR 1105 (HY000): vtgate: http://deepthi-ThinkPad:15001/: target: commerce.0.replica: vttablet: Connection Closed
+	//mysql> use @replica;
+	//Database changed
+	//mysql> select id from customer;
+	//ERROR 1105 (HY000): vtgate: http://deepthi-ThinkPad:15001/: target: commerce.0.replica: vttablet: Connection Closed
+
+	// begin transaction on replica
+	exec(t, replicaConn, "begin", "")
+	// try to delete a row, should fail
+	exec(t, replicaConn, "delete from customer where id=1", "supported only for master tablet type, current type: replica")
+	exec(t, replicaConn, "commit", "")
+
+	// begin transaction on replica
+	exec(t, replicaConn, "begin", "")
+	// try to update a row, should fail
+	exec(t, replicaConn, "update customer set email='emailn' where id=1", "supported only for master tablet type, current type: replica")
+	exec(t, replicaConn, "commit", "")
+
+	// begin transaction on replica
+	exec(t, replicaConn, "begin", "")
+	// try to insert a row, should fail
+	exec(t, replicaConn, "insert into customer(id, email) values(1,'email1')", "supported only for master tablet type, current type: replica")
+	// call rollback just for fun
+	exec(t, replicaConn, "rollback", "")
+}
+
 func getMapFromJSON(JSON map[string]interface{}, key string) map[string]interface{} {
 	result := make(map[string]interface{})
 	object := reflect.ValueOf(JSON[key])
@@ -99,9 +178,14 @@ func isMasterTabletPresent(tablets map[string]interface{}) bool {
 	return false
 }
 
-func exec(t *testing.T, conn *mysql.Conn, query string) *sqltypes.Result {
+func exec(t *testing.T, conn *mysql.Conn, query string, expectError string) *sqltypes.Result {
 	t.Helper()
 	qr, err := conn.ExecuteFetch(query, 1000, true)
-	require.Nil(t, err)
+	if expectError == "" {
+		require.Nil(t, err)
+	} else {
+		require.NotNil(t, err, "error should not be nil")
+		assert.Contains(t, err.Error(), expectError, "Unexpected error")
+	}
 	return qr
 }

go/vt/discovery/fake_healthcheck.go

@@ -111,8 +111,8 @@ func (fhc *FakeHealthCheck) ReplaceTablet(old, new *topodatapb.Tablet) {
 	fhc.AddTablet(new)
 }
 
-// GetConnection returns the TabletConn of the given tablet.
-func (fhc *FakeHealthCheck) GetConnection(key string) queryservice.QueryService {
+// TabletConnection returns the TabletConn of the given tablet.
+func (fhc *FakeHealthCheck) TabletConnection(key string) queryservice.QueryService {
 	fhc.mu.RLock()
 	defer fhc.mu.RUnlock()
 	if item := fhc.items[key]; item != nil {