test: added test to check binlogs to contain the cascade events (#13970)

Signed-off-by: Harshit Gangal <[email protected]>

go/test/endtoend/vtgate/foreignkey/fk_test.go

@@ -17,12 +17,19 @@ limitations under the License.
 package foreignkey
 
 import (
+	"context"
+	"io"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/utils"
+	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	"vitess.io/vitess/go/vt/vtgate/vtgateconn"
 )
 
 // TestInsertWithFK tests that insertions work as expected when foreign key management is enabled in Vitess.
@@ -101,7 +108,7 @@ func TestDeleteWithFK(t *testing.T) {
 	utils.AssertMatches(t, conn, `select * from u_t2`, `[[INT64(342) NULL] [INT64(19) INT64(1234)]]`)
 }
 
-// TestUpdations tests that update work as expected when foreign key management is enabled in Vitess.
+// TestUpdateWithFK tests that update work as expected when foreign key management is enabled in Vitess.
 func TestUpdateWithFK(t *testing.T) {
 	mcmp, closer := start(t)
 	conn := mcmp.VtConn
@@ -162,6 +169,125 @@ func TestUpdateWithFK(t *testing.T) {
 	utils.AssertMatches(t, conn, `select * from u_t3 order by id`, `[[INT64(1) INT64(12)] [INT64(32) INT64(13)]]`)
 }
 
+// TestVstreamForFKBinLog tests that dml queries with fks are written with child row first approach in the binary logs.
+func TestVstreamForFKBinLog(t *testing.T) {
+	vtgateConn, err := cluster.DialVTGate(context.Background(), t.Name(), vtgateGrpcAddress, "fk_user", "")
+	require.NoError(t, err)
+	defer vtgateConn.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	ch := make(chan *binlogdatapb.VEvent)
+	runVStream(t, ctx, ch, vtgateConn)
+
+	mcmp, closer := start(t)
+	conn := mcmp.VtConn
+	defer closer()
+	defer cancel()
+
+	utils.Exec(t, conn, `use uks`)
+
+	// insert some data.
+	utils.Exec(t, conn, `insert into u_t1(id, col1) values (1,2), (11,4), (111,6)`)
+	utils.Exec(t, conn, `insert into u_t2(id, col2) values (2,2), (22,4)`)
+	utils.Exec(t, conn, `insert into u_t3(id, col3) values (33,4), (333,6)`)
+	// drain 3 row events.
+	_ = drainEvents(t, ch, 3)
+
+	tcases := []struct {
+		query     string
+		events    int
+		rowEvents []string
+	}{{
+		query:  `update u_t1 set col1 = 3 where id = 11`,
+		events: 3,
+		rowEvents: []string{
+			`table_name:"uks.u_t3" row_changes:{before:{lengths:2 lengths:1 values:"334"} after:{lengths:2 lengths:1 values:"333"}} keyspace:"uks" shard:"0" flags:3`,
+			`table_name:"uks.u_t2" row_changes:{before:{lengths:2 lengths:1 values:"224"} after:{lengths:2 lengths:-1 values:"22"}} keyspace:"uks" shard:"0" flags:1`,
+			`table_name:"uks.u_t1" row_changes:{before:{lengths:2 lengths:1 values:"114"} after:{lengths:2 lengths:1 values:"113"}} keyspace:"uks" shard:"0" flags:1`,
+		},
+	}, {
+		query:  `update u_t1 set col1 = 5 where id = 11`,
+		events: 2,
+		rowEvents: []string{
+			`table_name:"uks.u_t3" row_changes:{before:{lengths:2 lengths:1 values:"333"} after:{lengths:2 lengths:1 values:"335"}} keyspace:"uks" shard:"0" flags:3`,
+			`table_name:"uks.u_t1" row_changes:{before:{lengths:2 lengths:1 values:"113"} after:{lengths:2 lengths:1 values:"115"}} keyspace:"uks" shard:"0" flags:1`,
+		},
+	}, {
+		query:  `delete from u_t1 where col1 = 6`,
+		events: 2,
+		rowEvents: []string{
+			`table_name:"uks.u_t3" row_changes:{before:{lengths:3 lengths:1 values:"3336"}} keyspace:"uks" shard:"0" flags:1`,
+			`table_name:"uks.u_t1" row_changes:{before:{lengths:3 lengths:1 values:"1116"}} keyspace:"uks" shard:"0" flags:1`,
+		},
+	}, {
+		query:  `update u_t1 set col1 = null where id = 11`,
+		events: 2,
+		rowEvents: []string{
+			`table_name:"uks.u_t3" row_changes:{before:{lengths:2 lengths:1 values:"335"} after:{lengths:2 lengths:-1 values:"33"}} keyspace:"uks" shard:"0" flags:3`,
+			`table_name:"uks.u_t1" row_changes:{before:{lengths:2 lengths:1 values:"115"} after:{lengths:2 lengths:-1 values:"11"}} keyspace:"uks" shard:"0" flags:1`,
+		},
+	}, {
+		query:  `delete from u_t1 where id = 11`,
+		events: 1,
+		rowEvents: []string{
+			`table_name:"uks.u_t1" row_changes:{before:{lengths:2 lengths:-1 values:"11"}} keyspace:"uks" shard:"0" flags:1`,
+		},
+	}}
+	for _, tcase := range tcases {
+		t.Run(tcase.query, func(t *testing.T) {
+			utils.Exec(t, conn, tcase.query)
+			// drain row events.
+			rowEvents := drainEvents(t, ch, tcase.events)
+			assert.ElementsMatch(t, tcase.rowEvents, rowEvents)
+		})
+	}
+}
+
+func runVStream(t *testing.T, ctx context.Context, ch chan *binlogdatapb.VEvent, vtgateConn *vtgateconn.VTGateConn) {
+	vgtid := &binlogdatapb.VGtid{
+		ShardGtids: []*binlogdatapb.ShardGtid{
+			{Keyspace: unshardedKs, Shard: "0", Gtid: "current"},
+		}}
+	filter := &binlogdatapb.Filter{
+		Rules: []*binlogdatapb.Rule{{
+			Match: "/u.*",
+		}},
+	}
+	vReader, err := vtgateConn.VStream(ctx, topodatapb.TabletType_REPLICA, vgtid, filter, nil)
+	require.NoError(t, err)
+
+	go func() {
+		for {
+			evs, err := vReader.Recv()
+			if err == io.EOF || ctx.Err() != nil {
+				return
+			}
+			require.NoError(t, err)
+
+			for _, ev := range evs {
+				if ev.Type == binlogdatapb.VEventType_ROW {
+					ch <- ev
+				}
+			}
+		}
+	}()
+}
+
+func drainEvents(t *testing.T, ch chan *binlogdatapb.VEvent, count int) []string {
+	var rowEvents []string
+	for i := 0; i < count; i++ {
+		select {
+		case re := <-ch:
+			rowEvents = append(rowEvents, re.RowEvent.String())
+		case <-time.After(10 * time.Second):
+			t.Fatalf("timeout waiting for event number: %d", i+1)
+		}
+	}
+	return rowEvents
+}
+
 // TestFkScenarios tests the various foreign key scenarios with different constraints
 // and makes sure that Vitess works with them as expected. All the tables are present in both sharded and unsharded keyspace
 // and all the foreign key constraints are cross-shard ones for the sharded keyspace.

go/test/endtoend/vtgate/foreignkey/main_test.go

@@ -25,19 +25,19 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"vitess.io/vitess/go/test/endtoend/utils"
-
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/test/endtoend/cluster"
+	"vitess.io/vitess/go/test/endtoend/utils"
 )
 
 var (
-	clusterInstance *cluster.LocalProcessCluster
-	vtParams        mysql.ConnParams
-	mysqlParams     mysql.ConnParams
-	shardedKs       = "ks"
-	unshardedKs     = "uks"
-	Cell            = "test"
+	clusterInstance   *cluster.LocalProcessCluster
+	vtParams          mysql.ConnParams
+	mysqlParams       mysql.ConnParams
+	vtgateGrpcAddress string
+	shardedKs         = "ks"
+	unshardedKs       = "uks"
+	Cell              = "test"
 	//go:embed sharded_schema.sql
 	shardedSchemaSQL string
 
@@ -82,7 +82,7 @@ func TestMain(m *testing.M) {
 			SchemaSQL: unshardedSchemaSQL,
 			VSchema:   unshardedVSchema,
 		}
-		err = clusterInstance.StartUnshardedKeyspace(*uKs, 0, false)
+		err = clusterInstance.StartUnshardedKeyspace(*uKs, 1, false)
 		if err != nil {
 			return 1
 		}
@@ -101,6 +101,7 @@ func TestMain(m *testing.M) {
 			Host: clusterInstance.Hostname,
 			Port: clusterInstance.VtgateMySQLPort,
 		}
+		vtgateGrpcAddress = fmt.Sprintf("%s:%d", clusterInstance.Hostname, clusterInstance.VtgateGrpcPort)
 
 		connParams, closer, err := utils.NewMySQL(clusterInstance, shardedKs, shardedSchemaSQL)
 		if err != nil {