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/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,80 @@ 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")
+
+	// 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
+
+	//mysql> use @replica;
+	//Reading table information for completion of table and column names
+	//You can turn off this feature to get a quicker startup with -A
+	//
+	//Database changed
+	//mysql> begin;
+	//Query OK, 0 rows affected (0.00 sec)
+	// mysql> insert into customer (email) values('f@d.com');
+	// ERROR 1290 (HY000): vtgate: http://deepthi-ThinkPad:15001/: execInsertUnsharded: vttablet: rpc error: code = FailedPrecondition desc = The MySQL server is running with the --read-only option so it cannot execute this statement (errno 1290) (sqlstate HY000) (CallerID: userData1): Sql: "insert into customer(email) values (:vtg1)", BindVars: {vtg1: "type:VARBINARY value:\"f@d.com\" "}
+	// TODO(deepthi): expecting a different error here:
+	//
+}
+
 func getMapFromJSON(JSON map[string]interface{}, key string) map[string]interface{} {
 	result := make(map[string]interface{})
 	object := reflect.ValueOf(JSON[key])

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 {