[release-19.0] Fix regression where inserts into reference tables with a different name on sharded keyspaces were not routed correctly. (#15796)

go/test/endtoend/vreplication/reference_test.go

@@ -0,0 +1,166 @@
+/*
+Copyright 2024 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package vreplication
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	uksSchema = `
+create table product (id int, mfg_id int, cat_id int, name varchar(128), primary key(id));
+create table cat (id int, name varchar(128), primary key(id));
+create table mfg (id int, name varchar(128), primary key(id));
+`
+	sksSchema = `
+create table product (id int, mfg_id int, cat_id int, name varchar(128), primary key(id));
+create table cat (id int, name varchar(128), primary key(id));
+create table mfg2 (id int, name varchar(128), primary key(id));
+`
+	uksVSchema = `
+{
+	"sharded": false,
+	"tables": {
+		"product": {},
+		"cat": {},
+		"mfg": {}
+	}
+}`
+
+	sksVSchema = `
+{	
+	"sharded": true,
+	"tables": {
+		"product": {
+			"column_vindexes": [
+				{	
+					"column": "id",	
+					"name": "hash"
+				}
+			]
+		},
+  		"cat": {
+			"type": "reference",
+      		"source": "uks.cat"
+		},
+		"mfg2": {
+			"type": "reference",
+		  		"source": "uks.mfg"
+		}
+	},
+	"vindexes": {
+		"hash": {
+			"type": "hash"
+		}
+	}
+}`
+	materializeCatSpec = `
+{
+	"workflow": "wfCat", 
+	"source_keyspace": "uks", 
+	"target_keyspace": "sks", 
+	"table_settings": [ {"target_table": "cat", "source_expression": "select id, name from cat"  }] 
+}`
+	materializeMfgSpec = `
+{
+	"workflow": "wfMfg",
+	"source_keyspace": "uks",
+	"target_keyspace": "sks",
+	"table_settings": [ {"target_table": "mfg2", "source_expression": "select id, name from mfg"  }] 
+}`
+	initializeTables = `
+use uks;
+insert into product values (1, 1, 1, 'p1');
+insert into product values (2, 2, 2, 'p2');
+insert into product values (3, 3, 3, 'p3');
+insert into cat values (1, 'c1');
+insert into cat values (2, 'c2');
+insert into cat values (3, 'c3');
+insert into mfg values (1, 'm1');
+insert into mfg values (2, 'm2');
+insert into mfg values (3, 'm3');
+insert into mfg values (4, 'm4');
+`
+)
+
+func TestReferenceTableMaterializationAndRouting(t *testing.T) {
+	var err error
+	defaultCellName := "zone1"
+	vc = NewVitessCluster(t, nil)
+	defer vc.TearDown()
+	defaultReplicas = 0 // because of CI resource constraints we can only run this test with primary tablets
+	defer func() { defaultReplicas = 1 }()
+	uks := "uks"
+	sks := "sks"
+
+	defaultCell := vc.Cells[defaultCellName]
+	vc.AddKeyspace(t, []*Cell{defaultCell}, uks, "0", uksVSchema, uksSchema, defaultReplicas, defaultRdonly, 100, nil)
+	vc.AddKeyspace(t, []*Cell{defaultCell}, sks, "-80,80-", sksVSchema, sksSchema, defaultReplicas, defaultRdonly, 200, nil)
+	vtgateConn := getConnection(t, vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateMySQLPort)
+
+	verifyClusterHealth(t, vc)
+	_, _, err = vtgateConn.ExecuteFetchMulti(initializeTables, 0, false)
+	require.NoError(t, err)
+	vtgateConn.Close()
+
+	materialize(t, materializeCatSpec, false)
+	materialize(t, materializeMfgSpec, false)
+
+	tabDash80 := vc.getPrimaryTablet(t, sks, "-80")
+	tab80Dash := vc.getPrimaryTablet(t, sks, "80-")
+	catchup(t, tabDash80, "wfCat", "Materialize Category")
+	catchup(t, tab80Dash, "wfCat", "Materialize Category")
+	catchup(t, tabDash80, "wfMfg", "Materialize Manufacturer")
+	catchup(t, tab80Dash, "wfMfg", "Materialize Manufacturer")
+
+	vtgateConn = getConnection(t, vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateMySQLPort)
+	defer vtgateConn.Close()
+	waitForRowCount(t, vtgateConn, sks, "cat", 3)
+	waitForRowCount(t, vtgateConn, sks, "mfg2", 4)
+
+	execRefQuery(t, "insert into mfg values (5, 'm5')")
+	execRefQuery(t, "insert into mfg2 values (6, 'm6')")
+	execRefQuery(t, "insert into uks.mfg values (7, 'm7')")
+	execRefQuery(t, "insert into sks.mfg2 values (8, 'm8')")
+	waitForRowCount(t, vtgateConn, uks, "mfg", 8)
+
+	execRefQuery(t, "update mfg set name = concat(name, '-updated') where id = 1")
+	execRefQuery(t, "update mfg2 set name = concat(name, '-updated') where id = 2")
+	execRefQuery(t, "update uks.mfg set name = concat(name, '-updated') where id = 3")
+	execRefQuery(t, "update sks.mfg2 set name = concat(name, '-updated') where id = 4")
+
+	waitForRowCount(t, vtgateConn, uks, "mfg", 8)
+	qr := execVtgateQuery(t, vtgateConn, "uks", "select count(*) from uks.mfg where name like '%updated%'")
+	require.NotNil(t, qr)
+	require.Equal(t, "4", qr.Rows[0][0].ToString())
+
+	execRefQuery(t, "delete from mfg where id = 5")
+	execRefQuery(t, "delete from mfg2 where id = 6")
+	execRefQuery(t, "delete from uks.mfg where id = 7")
+	execRefQuery(t, "delete from sks.mfg2 where id = 8")
+	waitForRowCount(t, vtgateConn, uks, "mfg", 4)
+
+}
+
+func execRefQuery(t *testing.T, query string) {
+	vtgateConn := getConnection(t, vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateMySQLPort)
+	defer vtgateConn.Close()
+	_, err := vtgateConn.ExecuteFetch(query, 0, false)
+	require.NoError(t, err)
+}

go/vt/vtgate/planbuilder/operators/insert.go

@@ -107,7 +107,7 @@ func (i *Insert) Statement() sqlparser.Statement {
 func createOperatorFromInsert(ctx *plancontext.PlanningContext, ins *sqlparser.Insert) Operator {
 	tableInfo, qt := createQueryTableForDML(ctx, ins.Table, nil)
 
-	vTbl, routing := buildVindexTableForDML(ctx, tableInfo, qt, "insert")
+	vTbl, routing := buildVindexTableForDML(ctx, tableInfo, qt, ins, "insert")
 
 	deleteBeforeInsert := false
 	if ins.Action == sqlparser.ReplaceAct &&

go/vt/vtgate/planbuilder/operators/route_planning.go

@@ -80,15 +80,25 @@
 	ctx *plancontext.PlanningContext,
 	tableInfo semantics.TableInfo,
 	table *QueryTable,
+	stmt sqlparser.Statement,
 	dmlType string,
 ) (*vindexes.Table, Routing) {
 	vindexTable := tableInfo.GetVindexTable()
-	if vindexTable.Source != nil {
+	if tableInfo.GetVindexTable().Type == vindexes.TypeReference && vindexTable.Source != nil {
 		sourceTable, _, _, _, _, err := ctx.VSchema.FindTableOrVindex(vindexTable.Source.TableName)
 		if err != nil {
 			panic(err)
 		}
 		vindexTable = sourceTable
+		refTbl := sqlparser.NewAliasedTableExpr(vindexTable.GetTableName(), "")
+		switch stmt := stmt.(type) {
+		case *sqlparser.Update:
+			stmt.TableExprs[0] = refTbl
+		case *sqlparser.Insert:
+			stmt.Table = refTbl
+		default:
+			panic("unsupported DML type in buildVindexTableForDML")
+		}
 	}
 
 	if !vindexTable.Keyspace.Sharded {

go/vt/vtgate/planbuilder/operators/update.go

@@ -102,8 +102,16 @@
 func createOperatorFromUpdate(ctx *plancontext.PlanningContext, updStmt *sqlparser.Update) Operator {
 	tableInfo, qt := createQueryTableForDML(ctx, updStmt.TableExprs[0], updStmt.Where)
 
-	vindexTable, routing := buildVindexTableForDML(ctx, tableInfo, qt, "update")
+	vindexTable, routing := buildVindexTableForDML(ctx, tableInfo, qt, updStmt, "update")
 
+	if tableInfo.GetVindexTable().GetTableName() != vindexTable.GetTableName() {
+		name := vindexTable.Name.String()
+		refTable := sqlparser.NewIdentifierCS(name)
+		if !qt.Alias.As.IsEmpty() {
+			qt.Alias.As = refTable
+		}
+		qt.Table.Name = refTable
+	}
 	updClone := sqlparser.CloneRefOfUpdate(updStmt)
 	updOp := createUpdateOperator(ctx, updStmt, vindexTable, qt, routing)