Foreign key on update action with non literal values

go/test/endtoend/vtgate/foreignkey/fk_fuzz_test.go

@@ -28,9 +28,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/stretchr/testify/require"
 
-	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/utils"
 	"vitess.io/vitess/go/vt/log"
 )
@@ -131,41 +129,33 @@ func (fz *fuzzer) generateInsertDMLQuery() string {
 	if tableName == "fk_t20" {
 		colValue := rand.Intn(1 + fz.maxValForCol)
 		col2Value := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("insert into %v (id, col, col2) values (%v, %v, %v)", tableName, idValue, convertColValueToString(colValue), convertColValueToString(col2Value))
+		return fmt.Sprintf("insert into %v (id, col, col2) values (%v, %v, %v)", tableName, idValue, convertIntValueToString(colValue), convertIntValueToString(col2Value))
 	} else if isMultiColFkTable(tableName) {
 		colaValue := rand.Intn(1 + fz.maxValForCol)
 		colbValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("insert into %v (id, cola, colb) values (%v, %v, %v)", tableName, idValue, convertColValueToString(colaValue), convertColValueToString(colbValue))
+		return fmt.Sprintf("insert into %v (id, cola, colb) values (%v, %v, %v)", tableName, idValue, convertIntValueToString(colaValue), convertIntValueToString(colbValue))
 	} else {
 		colValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("insert into %v (id, col) values (%v, %v)", tableName, idValue, convertColValueToString(colValue))
+		return fmt.Sprintf("insert into %v (id, col) values (%v, %v)", tableName, idValue, convertIntValueToString(colValue))
 	}
 }
 
-// convertColValueToString converts the given value to a string
-func convertColValueToString(value int) string {
-	if value == 0 {
-		return "NULL"
-	}
-	return fmt.Sprintf("%d", value)
-}
-
 // generateUpdateDMLQuery generates an UPDATE query from the parameters for the fuzzer.
 func (fz *fuzzer) generateUpdateDMLQuery() string {
 	tableId := rand.Intn(len(fkTables))
 	idValue := 1 + rand.Intn(fz.maxValForId)
 	tableName := fkTables[tableId]
 	if tableName == "fk_t20" {
-		colValue := rand.Intn(1 + fz.maxValForCol)
-		col2Value := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("update %v set col = %v, col2 = %v where id = %v", tableName, convertColValueToString(colValue), convertColValueToString(col2Value), idValue)
+		colValue := fz.generateExpression(rand.Intn(4)+1, "col", "col2")
+		col2Value := fz.generateExpression(rand.Intn(4)+1, "col", "col2")
+		return fmt.Sprintf("update %v set col = %v, col2 = %v where id = %v", tableName, colValue, col2Value, idValue)
 	} else if isMultiColFkTable(tableName) {
-		colaValue := rand.Intn(1 + fz.maxValForCol)
-		colbValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("update %v set cola = %v, colb = %v where id = %v", tableName, convertColValueToString(colaValue), convertColValueToString(colbValue), idValue)
+		colaValue := fz.generateExpression(rand.Intn(4)+1, "cola", "colb")
+		colbValue := fz.generateExpression(rand.Intn(4)+1, "cola", "colb")
+		return fmt.Sprintf("update %v set cola = %v, colb = %v where id = %v", tableName, colaValue, colbValue, idValue)
 	} else {
 		colValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("update %v set col = %v where id = %v", tableName, convertColValueToString(colValue), idValue)
+		return fmt.Sprintf("update %v set col = %v where id = %v", tableName, convertIntValueToString(colValue), idValue)
 	}
 }
 
@@ -338,8 +328,8 @@ func (fz *fuzzer) getPreparedInsertQueries() []string {
 		return []string{
 			"prepare stmt_insert from 'insert into fk_t20 (id, col, col2) values (?, ?, ?)'",
 			fmt.Sprintf("SET @id = %v", idValue),
-			fmt.Sprintf("SET @col = %v", convertColValueToString(colValue)),
-			fmt.Sprintf("SET @col2 = %v", convertColValueToString(col2Value)),
+			fmt.Sprintf("SET @col = %v", convertIntValueToString(colValue)),
+			fmt.Sprintf("SET @col2 = %v", convertIntValueToString(col2Value)),
 			"execute stmt_insert using @id, @col, @col2",
 		}
 	} else if isMultiColFkTable(tableName) {
@@ -348,16 +338,16 @@ func (fz *fuzzer) getPreparedInsertQueries() []string {
 		return []string{
 			fmt.Sprintf("prepare stmt_insert from 'insert into %v (id, cola, colb) values (?, ?, ?)'", tableName),
 			fmt.Sprintf("SET @id = %v", idValue),
-			fmt.Sprintf("SET @cola = %v", convertColValueToString(colaValue)),
-			fmt.Sprintf("SET @colb = %v", convertColValueToString(colbValue)),
+			fmt.Sprintf("SET @cola = %v", convertIntValueToString(colaValue)),
+			fmt.Sprintf("SET @colb = %v", convertIntValueToString(colbValue)),
 			"execute stmt_insert using @id, @cola, @colb",
 		}
 	} else {
 		colValue := rand.Intn(1 + fz.maxValForCol)
 		return []string{
 			fmt.Sprintf("prepare stmt_insert from 'insert into %v (id, col) values (?, ?)'", tableName),
 			fmt.Sprintf("SET @id = %v", idValue),
-			fmt.Sprintf("SET @col = %v", convertColValueToString(colValue)),
+			fmt.Sprintf("SET @col = %v", convertIntValueToString(colValue)),
 			"execute stmt_insert using @id, @col",
 		}
 	}
@@ -374,8 +364,8 @@ func (fz *fuzzer) getPreparedUpdateQueries() []string {
 		return []string{
 			"prepare stmt_update from 'update fk_t20 set col = ?, col2 = ? where id = ?'",
 			fmt.Sprintf("SET @id = %v", idValue),
-			fmt.Sprintf("SET @col = %v", convertColValueToString(colValue)),
-			fmt.Sprintf("SET @col2 = %v", convertColValueToString(col2Value)),
+			fmt.Sprintf("SET @col = %v", convertIntValueToString(colValue)),
+			fmt.Sprintf("SET @col2 = %v", convertIntValueToString(col2Value)),
 			"execute stmt_update using @col, @col2, @id",
 		}
 	} else if isMultiColFkTable(tableName) {
@@ -384,16 +374,16 @@ func (fz *fuzzer) getPreparedUpdateQueries() []string {
 		return []string{
 			fmt.Sprintf("prepare stmt_update from 'update %v set cola = ?, colb = ? where id = ?'", tableName),
 			fmt.Sprintf("SET @id = %v", idValue),
-			fmt.Sprintf("SET @cola = %v", convertColValueToString(colaValue)),
-			fmt.Sprintf("SET @colb = %v", convertColValueToString(colbValue)),
+			fmt.Sprintf("SET @cola = %v", convertIntValueToString(colaValue)),
+			fmt.Sprintf("SET @colb = %v", convertIntValueToString(colbValue)),
 			"execute stmt_update using @cola, @colb, @id",
 		}
 	} else {
 		colValue := rand.Intn(1 + fz.maxValForCol)
 		return []string{
 			fmt.Sprintf("prepare stmt_update from 'update %v set col = ? where id = ?'", tableName),
 			fmt.Sprintf("SET @id = %v", idValue),
-			fmt.Sprintf("SET @col = %v", convertColValueToString(colValue)),
+			fmt.Sprintf("SET @col = %v", convertIntValueToString(colValue)),
 			"execute stmt_update using @col, @id",
 		}
 	}
@@ -419,14 +409,14 @@ func (fz *fuzzer) generateParameterizedInsertQuery() (query string, params []any
 	if tableName == "fk_t20" {
 		colValue := rand.Intn(1 + fz.maxValForCol)
 		col2Value := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("insert into %v (id, col, col2) values (?, ?, ?)", tableName), []any{idValue, convertColValueToString(colValue), convertColValueToString(col2Value)}
+		return fmt.Sprintf("insert into %v (id, col, col2) values (?, ?, ?)", tableName), []any{idValue, convertIntValueToString(colValue), convertIntValueToString(col2Value)}
 	} else if isMultiColFkTable(tableName) {
 		colaValue := rand.Intn(1 + fz.maxValForCol)
 		colbValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("insert into %v (id, cola, colb) values (?, ?, ?)", tableName), []any{idValue, convertColValueToString(colaValue), convertColValueToString(colbValue)}
+		return fmt.Sprintf("insert into %v (id, cola, colb) values (?, ?, ?)", tableName), []any{idValue, convertIntValueToString(colaValue), convertIntValueToString(colbValue)}
 	} else {
 		colValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("insert into %v (id, col) values (?, ?)", tableName), []any{idValue, convertColValueToString(colValue)}
+		return fmt.Sprintf("insert into %v (id, col) values (?, ?)", tableName), []any{idValue, convertIntValueToString(colValue)}
 	}
 }
 
@@ -438,14 +428,14 @@ func (fz *fuzzer) generateParameterizedUpdateQuery() (query string, params []any
 	if tableName == "fk_t20" {
 		colValue := rand.Intn(1 + fz.maxValForCol)
 		col2Value := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("update %v set col = ?, col2 = ? where id = ?", tableName), []any{convertColValueToString(colValue), convertColValueToString(col2Value), idValue}
+		return fmt.Sprintf("update %v set col = ?, col2 = ? where id = ?", tableName), []any{convertIntValueToString(colValue), convertIntValueToString(col2Value), idValue}
 	} else if isMultiColFkTable(tableName) {
 		colaValue := rand.Intn(1 + fz.maxValForCol)
 		colbValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("update %v set cola = ?, colb = ? where id = ?", tableName), []any{convertColValueToString(colaValue), convertColValueToString(colbValue), idValue}
+		return fmt.Sprintf("update %v set cola = ?, colb = ? where id = ?", tableName), []any{convertIntValueToString(colaValue), convertIntValueToString(colbValue), idValue}
 	} else {
 		colValue := rand.Intn(1 + fz.maxValForCol)
-		return fmt.Sprintf("update %v set col = ? where id = ?", tableName), []any{convertColValueToString(colValue), idValue}
+		return fmt.Sprintf("update %v set col = ? where id = ?", tableName), []any{convertIntValueToString(colValue), idValue}
 	}
 }
 
@@ -650,85 +640,3 @@ func TestFkFuzzTest(t *testing.T) {
 		}
 	}
 }
-
-// ensureDatabaseState ensures that the database is either empty or not.
-func ensureDatabaseState(t *testing.T, vtconn *mysql.Conn, empty bool) {
-	results := collectFkTablesState(vtconn)
-	isEmpty := true
-	for _, res := range results {
-		if len(res.Rows) > 0 {
-			isEmpty = false
-		}
-	}
-	require.Equal(t, isEmpty, empty)
-}
-
-// verifyDataIsCorrect verifies that the data in MySQL database matches the data in the Vitess database.
-func verifyDataIsCorrect(t *testing.T, mcmp utils.MySQLCompare, concurrency int) {
-	// For single concurrent thread, we run all the queries on both MySQL and Vitess, so we can verify correctness
-	// by just checking if the data in MySQL and Vitess match.
-	if concurrency == 1 {
-		for _, table := range fkTables {
-			query := fmt.Sprintf("SELECT * FROM %v ORDER BY id", table)
-			mcmp.Exec(query)
-		}
-	} else {
-		// For higher concurrency, we don't have MySQL data to verify everything is fine,
-		// so we'll have to do something different.
-		// We run LEFT JOIN queries on all the parent and child tables linked by foreign keys
-		// to make sure that nothing is broken in the database.
-		for _, reference := range fkReferences {
-			query := fmt.Sprintf("select %v.id from %v left join %v on (%v.col = %v.col) where %v.col is null and %v.col is not null", reference.childTable, reference.childTable, reference.parentTable, reference.parentTable, reference.childTable, reference.parentTable, reference.childTable)
-			if isMultiColFkTable(reference.childTable) {
-				query = fmt.Sprintf("select %v.id from %v left join %v on (%v.cola = %v.cola and %v.colb = %v.colb) where %v.cola is null and %v.cola is not null and %v.colb is not null", reference.childTable, reference.childTable, reference.parentTable, reference.parentTable, reference.childTable, reference.parentTable, reference.childTable, reference.parentTable, reference.childTable, reference.childTable)
-			}
-			res, err := mcmp.VtConn.ExecuteFetch(query, 1000, false)
-			require.NoError(t, err)
-			require.Zerof(t, len(res.Rows), "Query %v gave non-empty results", query)
-		}
-	}
-	// We also verify that the results in Primary and Replica table match as is.
-	for _, keyspace := range clusterInstance.Keyspaces {
-		for _, shard := range keyspace.Shards {
-			var primaryTab, replicaTab *cluster.Vttablet
-			for _, vttablet := range shard.Vttablets {
-				if vttablet.Type == "primary" {
-					primaryTab = vttablet
-				} else {
-					replicaTab = vttablet
-				}
-			}
-			require.NotNil(t, primaryTab)
-			require.NotNil(t, replicaTab)
-			checkReplicationHealthy(t, replicaTab)
-			cluster.WaitForReplicationPos(t, primaryTab, replicaTab, true, 60.0)
-			primaryConn, err := utils.GetMySQLConn(primaryTab, fmt.Sprintf("vt_%v", keyspace.Name))
-			require.NoError(t, err)
-			replicaConn, err := utils.GetMySQLConn(replicaTab, fmt.Sprintf("vt_%v", keyspace.Name))
-			require.NoError(t, err)
-			primaryRes := collectFkTablesState(primaryConn)
-			replicaRes := collectFkTablesState(replicaConn)
-			verifyDataMatches(t, primaryRes, replicaRes)
-		}
-	}
-}
-
-// verifyDataMatches verifies that the two list of results are the same.
-func verifyDataMatches(t *testing.T, resOne []*sqltypes.Result, resTwo []*sqltypes.Result) {
-	require.EqualValues(t, len(resTwo), len(resOne), "Res 1 - %v, Res 2 - %v", resOne, resTwo)
-	for idx, resultOne := range resOne {
-		resultTwo := resTwo[idx]
-		require.True(t, resultOne.Equal(resultTwo), "Data for %v doesn't match\nRows 1\n%v\nRows 2\n%v", fkTables[idx], resultOne.Rows, resultTwo.Rows)
-	}
-}
-
-// collectFkTablesState collects the data stored in the foreign key tables for the given connection.
-func collectFkTablesState(conn *mysql.Conn) []*sqltypes.Result {
-	var tablesData []*sqltypes.Result
-	for _, table := range fkTables {
-		query := fmt.Sprintf("SELECT * FROM %v ORDER BY id", table)
-		res, _ := conn.ExecuteFetch(query, 10000, true)
-		tablesData = append(tablesData, res)
-	}
-	return tablesData
-}

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

@@ -27,6 +27,7 @@ import (
 
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/utils"
+	"vitess.io/vitess/go/vt/log"
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/vtgate/vtgateconn"
@@ -774,3 +775,133 @@ func TestFkScenarios(t *testing.T) {
 		})
 	}
 }
+
+// TestFkQueries is for testing a specific set of queries one after the other.
+func TestFkQueries(t *testing.T) {
+	// Wait for schema-tracking to be complete.
+	waitForSchemaTrackingForFkTables(t)
+	// Remove all the foreign key constraints for all the replicas.
+	// We can then verify that the replica, and the primary have the same data, to ensure
+	// that none of the queries ever lead to cascades/updates on MySQL level.
+	for _, ks := range []string{shardedKs, unshardedKs} {
+		replicas := getReplicaTablets(ks)
+		for _, replica := range replicas {
+			removeAllForeignKeyConstraints(t, replica, ks)
+		}
+	}
+
+	testcases := []struct {
+		name    string
+		queries []string
+	}{
+		{
+			name: "Non-literal update",
+			queries: []string{
+				"insert into fk_t10 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t11 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"update fk_t10 set col = id + 3",
+			},
+		}, {
+			name: "Non-literal update with order by",
+			queries: []string{
+				"insert into fk_t10 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t11 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"update fk_t10 set col = id + 3 order by id desc",
+			},
+		}, {
+			name: "Non-literal update with order by that require parent and child foreign keys verification - success",
+			queries: []string{
+				"insert into fk_t10 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5),(6,6),(7,7),(8,8)",
+				"insert into fk_t11 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t12 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t13 (id, col) values (1,1),(2,2)",
+				"update fk_t11 set col = id + 3 where id >= 3",
+			},
+		}, {
+			name: "Non-literal update with order by that require parent and child foreign keys verification - parent fails",
+			queries: []string{
+				"insert into fk_t10 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t11 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t12 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"update fk_t11 set col = id + 3",
+			},
+		}, {
+			name: "Non-literal update with order by that require parent and child foreign keys verification - child fails",
+			queries: []string{
+				"insert into fk_t10 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5),(6,6),(7,7),(8,8)",
+				"insert into fk_t11 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t12 (id, col) values (1,1),(2,2),(3,3),(4,4),(5,5)",
+				"insert into fk_t13 (id, col) values (1,1),(2,2)",
+				"update fk_t11 set col = id + 3",
+			},
+		},
+	}
+
+	for _, testcase := range testcases {
+		t.Run(testcase.name, func(t *testing.T) {
+			mcmp, closer := start(t)
+			defer closer()
+			_ = utils.Exec(t, mcmp.VtConn, "use `uks`")
+
+			// Ensure that the Vitess database is originally empty
+			ensureDatabaseState(t, mcmp.VtConn, true)
+			ensureDatabaseState(t, mcmp.MySQLConn, true)
+
+			for _, query := range testcase.queries {
+				_, _ = mcmp.ExecAllowAndCompareError(query)
+				if t.Failed() {
+					break
+				}
+			}
+
+			// ensure Vitess database has some data. This ensures not all the commands failed.
+			ensureDatabaseState(t, mcmp.VtConn, false)
+			// Verify the consistency of the data.
+			verifyDataIsCorrect(t, mcmp, 1)
+		})
+	}
+}
+
+// TestFkOneCase is for testing a specific set of queries. On the CI this test won't run since we'll keep the queries empty.
+func TestFkOneCase(t *testing.T) {
+	queries := []string{}
+	if len(queries) == 0 {
+		t.Skip("No queries to test")
+	}
+	// Wait for schema-tracking to be complete.
+	waitForSchemaTrackingForFkTables(t)
+	// Remove all the foreign key constraints for all the replicas.
+	// We can then verify that the replica, and the primary have the same data, to ensure
+	// that none of the queries ever lead to cascades/updates on MySQL level.
+	for _, ks := range []string{shardedKs, unshardedKs} {
+		replicas := getReplicaTablets(ks)
+		for _, replica := range replicas {
+			removeAllForeignKeyConstraints(t, replica, ks)
+		}
+	}
+
+	mcmp, closer := start(t)
+	defer closer()
+	_ = utils.Exec(t, mcmp.VtConn, "use `uks`")
+
+	// Ensure that the Vitess database is originally empty
+	ensureDatabaseState(t, mcmp.VtConn, true)
+	ensureDatabaseState(t, mcmp.MySQLConn, true)
+
+	for _, query := range queries {
+		_, _ = mcmp.ExecAllowAndCompareError(query)
+		if t.Failed() {
+			log.Errorf("Query failed - %v", query)
+			break
+		}
+	}
+	vitessData := collectFkTablesState(mcmp.VtConn)
+	for idx, table := range fkTables {
+		log.Errorf("Vitess data for %v -\n%v", table, vitessData[idx].Rows)
+	}
+
+	// ensure Vitess database has some data. This ensures not all the commands failed.
+	ensureDatabaseState(t, mcmp.VtConn, false)
+	// Verify the consistency of the data.
+	verifyDataIsCorrect(t, mcmp, 1)
+}