diff --git a/go/mysql/sqlerror/sql_error.go b/go/mysql/sqlerror/sql_error.go
index 06b0ade9e40..9b1f65c82e3 100644
--- a/go/mysql/sqlerror/sql_error.go
+++ b/go/mysql/sqlerror/sql_error.go
@@ -213,6 +213,7 @@ var stateToMysqlCode = map[vterrors.State]mysqlCode{
 	vterrors.CantDoThisInTransaction:      {num: ERCantDoThisDuringAnTransaction, state: SSCantDoThisDuringAnTransaction},
 	vterrors.RequiresPrimaryKey:           {num: ERRequiresPrimaryKey, state: SSClientError},
 	vterrors.RowIsReferenced2:             {num: ERRowIsReferenced2, state: SSConstraintViolation},
+	vterrors.NoReferencedRow2:             {num: ErNoReferencedRow2, state: SSConstraintViolation},
 	vterrors.NoSuchSession:                {num: ERUnknownComError, state: SSNetError},
 	vterrors.OperandColumns:               {num: EROperandColumns, state: SSWrongNumberOfColumns},
 	vterrors.WrongValueCountOnRow:         {num: ERWrongValueCountOnRow, state: SSWrongValueCountOnRow},
diff --git a/go/vt/vterrors/state.go b/go/vt/vterrors/state.go
index ce2792e35f0..5e3dcf22dfb 100644
--- a/go/vt/vterrors/state.go
+++ b/go/vt/vterrors/state.go
@@ -56,6 +56,7 @@ const (
 	RequiresPrimaryKey
 	OperandColumns
 	RowIsReferenced2
+	NoReferencedRow2
 	UnknownStmtHandler
 
 	// not found
diff --git a/go/vt/vtgate/engine/cached_size.go b/go/vt/vtgate/engine/cached_size.go
index 838b6b5356b..8ef69bfb89e 100644
--- a/go/vt/vtgate/engine/cached_size.go
+++ b/go/vt/vtgate/engine/cached_size.go
@@ -294,6 +294,64 @@ func (cached *FkChild) CachedSize(alloc bool) int64 {
 	}
 	return size
 }
+func (cached *FkParent) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(80)
+	}
+	// field Values []vitess.io/vitess/go/vt/sqlparser.Exprs
+	{
+		size += hack.RuntimeAllocSize(int64(cap(cached.Values)) * int64(24))
+		for _, elem := range cached.Values {
+			{
+				size += hack.RuntimeAllocSize(int64(cap(elem)) * int64(16))
+				for _, elem := range elem {
+					if cc, ok := elem.(cachedObject); ok {
+						size += cc.CachedSize(true)
+					}
+				}
+			}
+		}
+	}
+	// field Cols []vitess.io/vitess/go/vt/vtgate/engine.CheckCol
+	{
+		size += hack.RuntimeAllocSize(int64(cap(cached.Cols)) * int64(22))
+		for _, elem := range cached.Cols {
+			size += elem.CachedSize(false)
+		}
+	}
+	// field BvName string
+	size += hack.RuntimeAllocSize(int64(len(cached.BvName)))
+	// field Exec vitess.io/vitess/go/vt/vtgate/engine.Primitive
+	if cc, ok := cached.Exec.(cachedObject); ok {
+		size += cc.CachedSize(true)
+	}
+	return size
+}
+func (cached *FkVerify) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(48)
+	}
+	// field Verify []*vitess.io/vitess/go/vt/vtgate/engine.FkParent
+	{
+		size += hack.RuntimeAllocSize(int64(cap(cached.Verify)) * int64(8))
+		for _, elem := range cached.Verify {
+			size += elem.CachedSize(true)
+		}
+	}
+	// field Exec vitess.io/vitess/go/vt/vtgate/engine.Primitive
+	if cc, ok := cached.Exec.(cachedObject); ok {
+		size += cc.CachedSize(true)
+	}
+	return size
+}
 func (cached *Generate) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
diff --git a/go/vt/vtgate/engine/fk_verify.go b/go/vt/vtgate/engine/fk_verify.go
new file mode 100644
index 00000000000..9f0c6b33da8
--- /dev/null
+++ b/go/vt/vtgate/engine/fk_verify.go
@@ -0,0 +1,195 @@
+/*
+Copyright 2023 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 engine
+
+import (
+	"context"
+
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vterrors"
+)
+
+// FkParent is a primitive that represents a parent table foreign key constraint to verify against.
+type FkParent struct {
+	Values []sqlparser.Exprs
+	Cols   []CheckCol
+	BvName string
+
+	Exec Primitive
+}
+
+// FkVerify is a primitive that verifies that the foreign key constraints in parent tables are satisfied.
+// It does this by executing a select distinct query on the parent table with the values that are being inserted/updated.
+type FkVerify struct {
+	Verify []*FkParent
+	Exec   Primitive
+
+	txNeeded
+}
+
+// RouteType implements the Primitive interface
+func (f *FkVerify) RouteType() string {
+	return "FKVerify"
+}
+
+// GetKeyspaceName implements the Primitive interface
+func (f *FkVerify) GetKeyspaceName() string {
+	return f.Exec.GetKeyspaceName()
+}
+
+// GetTableName implements the Primitive interface
+func (f *FkVerify) GetTableName() string {
+	return f.Exec.GetTableName()
+}
+
+// GetFields implements the Primitive interface
+func (f *FkVerify) GetFields(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG] GetFields should not be called")
+}
+
+// TryExecute implements the Primitive interface
+func (f *FkVerify) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
+	for _, fk := range f.Verify {
+		pt := newProbeTable(fk.Cols)
+		newBv := &querypb.BindVariable{
+			Type: querypb.Type_TUPLE,
+		}
+		for _, exprs := range fk.Values {
+			var row sqltypes.Row
+			var values []*querypb.Value
+			for _, expr := range exprs {
+				val, err := getValue(expr, bindVars)
+				if err != nil {
+					return nil, vterrors.Wrapf(err, "unable to get value for the expression %v", expr)
+				}
+				row = append(row, val)
+				values = append(values, sqltypes.ValueToProto(val))
+			}
+			if exists, err := pt.exists(row); err != nil {
+				return nil, err
+			} else if !exists {
+				newBv.Values = append(newBv.Values, &querypb.Value{Type: querypb.Type_TUPLE, Values: values})
+			}
+		}
+		distinctValues := len(newBv.Values)
+		qr, err := vcursor.ExecutePrimitive(ctx, fk.Exec, map[string]*querypb.BindVariable{fk.BvName: newBv}, wantfields)
+		if err != nil {
+			return nil, err
+		}
+		if distinctValues != len(qr.Rows) {
+			return nil, vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.NoReferencedRow2, "Cannot add or update a child row: a foreign key constraint fails")
+		}
+	}
+	return vcursor.ExecutePrimitive(ctx, f.Exec, bindVars, wantfields)
+}
+
+// TryStreamExecute implements the Primitive interface
+func (f *FkVerify) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	for _, fk := range f.Verify {
+		pt := newProbeTable(fk.Cols)
+		newBv := &querypb.BindVariable{
+			Type: querypb.Type_TUPLE,
+		}
+		for _, exprs := range fk.Values {
+			var row sqltypes.Row
+			var values []*querypb.Value
+			for _, expr := range exprs {
+				val, err := getValue(expr, bindVars)
+				if err != nil {
+					return vterrors.Wrapf(err, "unable to get value for the expression %v", expr)
+				}
+				row = append(row, val)
+				values = append(values, sqltypes.ValueToProto(val))
+			}
+			if exists, err := pt.exists(row); err != nil {
+				return err
+			} else if !exists {
+				newBv.Values = append(newBv.Values, &querypb.Value{Type: querypb.Type_TUPLE, Values: values})
+			}
+		}
+		distinctValues := len(newBv.Values)
+
+		seenRows := 0
+		err := vcursor.StreamExecutePrimitive(ctx, fk.Exec, map[string]*querypb.BindVariable{fk.BvName: newBv}, wantfields, func(qr *sqltypes.Result) error {
+			seenRows += len(qr.Rows)
+			return nil
+		})
+		if err != nil {
+			return err
+		}
+
+		if distinctValues != seenRows {
+			return vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.NoReferencedRow2, "Cannot add or update a child row: a foreign key constraint fails")
+		}
+	}
+	return vcursor.StreamExecutePrimitive(ctx, f.Exec, bindVars, wantfields, callback)
+}
+
+// Inputs implements the Primitive interface
+func (f *FkVerify) Inputs() []Primitive {
+	return nil
+}
+
+func (f *FkVerify) description() PrimitiveDescription {
+	var parentDesc []PrimitiveDescription
+	for _, parent := range f.Verify {
+		parentDesc = append(parentDesc, PrimitiveDescription{
+			OperatorType: "FkVerifyParent",
+			Inputs: []PrimitiveDescription{
+				PrimitiveToPlanDescription(parent.Exec),
+			},
+			Other: map[string]any{
+				"BvName": parent.BvName,
+				"Cols":   parent.Cols,
+			},
+		})
+	}
+	return PrimitiveDescription{
+		OperatorType: f.RouteType(),
+		Other: map[string]any{
+			"Parent": parentDesc,
+			"Child":  PrimitiveToPlanDescription(f.Exec),
+		},
+	}
+}
+
+var _ Primitive = (*FkVerify)(nil)
+
+func getValue(expr sqlparser.Expr, bindVars map[string]*querypb.BindVariable) (sqltypes.Value, error) {
+	switch e := expr.(type) {
+	case *sqlparser.Literal:
+		return sqlparser.LiteralToValue(e)
+	case sqlparser.BoolVal:
+		b := int32(0)
+		if e {
+			b = 1
+		}
+		return sqltypes.NewInt32(b), nil
+	case *sqlparser.NullVal:
+		return sqltypes.NULL, nil
+	case *sqlparser.Argument:
+		bv, exists := bindVars[e.Name]
+		if !exists {
+			return sqltypes.Value{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG] bind variable %s missing", e.Name)
+		}
+		return sqltypes.BindVariableToValue(bv)
+	}
+	return sqltypes.Value{}, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG] unexpected expression type: %T", expr)
+}
diff --git a/go/vt/vtgate/engine/fk_verify_test.go b/go/vt/vtgate/engine/fk_verify_test.go
new file mode 100644
index 00000000000..5821510f93d
--- /dev/null
+++ b/go/vt/vtgate/engine/fk_verify_test.go
@@ -0,0 +1,187 @@
+/*
+Copyright 2023 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 engine
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/mysql/collations"
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vtgate/vindexes"
+)
+
+func TestFKVerifyUpdate(t *testing.T) {
+	verifyP := &Route{
+		Query: "select distinct cola, colb from parent where (cola, colb) in ::__vals",
+		RoutingParameters: &RoutingParameters{
+			Opcode:   Unsharded,
+			Keyspace: &vindexes.Keyspace{Name: "ks"},
+		},
+	}
+	childP := &Update{
+		DML: &DML{
+			Query: "update child set cola = 1, colb = 'a' where foo = 48",
+			RoutingParameters: &RoutingParameters{
+				Opcode:   Unsharded,
+				Keyspace: &vindexes.Keyspace{Name: "ks"},
+			},
+		},
+	}
+	fkc := &FkVerify{
+		Verify: []*FkParent{
+			{
+				Values: []sqlparser.Exprs{{sqlparser.NewIntLiteral("1"), sqlparser.NewStrLiteral("a")}},
+				Cols: []CheckCol{
+					{Col: 0, Type: sqltypes.Int64, Collation: collations.CollationBinaryID},
+					{Col: 1, Type: sqltypes.VarChar, Collation: collations.CollationUtf8mb4ID},
+				},
+				BvName: "__vals",
+				Exec:   verifyP,
+			},
+		},
+		Exec: childP,
+	}
+
+	t.Run("foreign key verification success", func(t *testing.T) {
+		fakeRes := sqltypes.MakeTestResult(sqltypes.MakeTestFields("cola|colb", "int64|varchar"), "1|a")
+		vc := newDMLTestVCursor("0")
+		vc.results = []*sqltypes.Result{fakeRes}
+		_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
+		require.NoError(t, err)
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: select distinct cola, colb from parent where (cola, colb) in ::__vals {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}}} false false`,
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: update child set cola = 1, colb = 'a' where foo = 48 {} true true`,
+		})
+
+		vc.Rewind()
+		err = fkc.TryStreamExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true, func(result *sqltypes.Result) error { return nil })
+		require.NoError(t, err)
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`StreamExecuteMulti select distinct cola, colb from parent where (cola, colb) in ::__vals ks.0: {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}}} `,
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: update child set cola = 1, colb = 'a' where foo = 48 {} true true`,
+		})
+	})
+
+	t.Run("foreign key verification failure", func(t *testing.T) {
+		// No results from select, should cause the foreign key verification to fail.
+		fakeRes := sqltypes.MakeTestResult(sqltypes.MakeTestFields("cola|colb", "int64|varchar"))
+		vc := newDMLTestVCursor("0")
+		vc.results = []*sqltypes.Result{fakeRes}
+		_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
+		require.ErrorContains(t, err, "Cannot add or update a child row: a foreign key constraint fails")
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: select distinct cola, colb from parent where (cola, colb) in ::__vals {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}}} false false`,
+		})
+
+		vc.Rewind()
+		err = fkc.TryStreamExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true, func(result *sqltypes.Result) error { return nil })
+		require.ErrorContains(t, err, "Cannot add or update a child row: a foreign key constraint fails")
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`StreamExecuteMulti select distinct cola, colb from parent where (cola, colb) in ::__vals ks.0: {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}}} `,
+		})
+	})
+}
+
+// TestFKVerifyInsert tests the functionality of FkVerify Primitive to verify the validation of the foreign key constraints when executing an insert.
+func TestFKVerifyInsert(t *testing.T) {
+	verifyP := &Route{
+		Query: "select distinct cola, colb from parent where (cola, colb) in ::__vals",
+		RoutingParameters: &RoutingParameters{
+			Opcode:   Unsharded,
+			Keyspace: &vindexes.Keyspace{Name: "ks"},
+		},
+	}
+	childP := &Insert{
+		Opcode:   InsertUnsharded,
+		Query:    "insert into child (cola, colb, x) values (1, 'a', 1), (2, 'b', 2), (1, 'a', 3),",
+		Keyspace: &vindexes.Keyspace{Name: "ks"},
+	}
+	fkc := &FkVerify{
+		Verify: []*FkParent{
+			{
+				Values: []sqlparser.Exprs{
+					{sqlparser.NewIntLiteral("1"), sqlparser.NewStrLiteral("a")},
+					{sqlparser.NewIntLiteral("2"), sqlparser.NewStrLiteral("b")},
+					{sqlparser.NewIntLiteral("1"), sqlparser.NewStrLiteral("a")},
+				},
+				Cols: []CheckCol{
+					{Col: 0, Type: sqltypes.Int64, Collation: collations.CollationBinaryID},
+					{Col: 1, Type: sqltypes.VarChar, Collation: collations.CollationUtf8mb4ID},
+				},
+				BvName: "__vals",
+				Exec:   verifyP,
+			},
+		},
+		Exec: childP,
+	}
+
+	t.Run("foreign key verification success", func(t *testing.T) {
+		fakeRes := sqltypes.MakeTestResult(sqltypes.MakeTestFields("cola|colb", "int64|varchar"), "1|a", "2|b")
+		vc := newDMLTestVCursor("0")
+		vc.results = []*sqltypes.Result{fakeRes}
+		_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
+		require.NoError(t, err)
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: select distinct cola, colb from parent where (cola, colb) in ::__vals {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}} values:{type:TUPLE values:{type:INT64 value:"2"} values:{type:VARCHAR value:"b"}}} false false`,
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: insert into child (cola, colb, x) values (1, 'a', 1), (2, 'b', 2), (1, 'a', 3), {} true true`,
+		})
+
+		vc.Rewind()
+		err = fkc.TryStreamExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true, func(result *sqltypes.Result) error { return nil })
+		require.NoError(t, err)
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`StreamExecuteMulti select distinct cola, colb from parent where (cola, colb) in ::__vals ks.0: {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}} values:{type:TUPLE values:{type:INT64 value:"2"} values:{type:VARCHAR value:"b"}}} `,
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: insert into child (cola, colb, x) values (1, 'a', 1), (2, 'b', 2), (1, 'a', 3), {} true true`,
+		})
+	})
+
+	t.Run("foreign key verification failure", func(t *testing.T) {
+		// Only 1 result from select, should cause the foreign key verification to fail.
+		fakeRes := sqltypes.MakeTestResult(sqltypes.MakeTestFields("cola|colb", "int64|varchar"), "2|b")
+		vc := newDMLTestVCursor("0")
+		vc.results = []*sqltypes.Result{fakeRes}
+		_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
+		require.ErrorContains(t, err, "Cannot add or update a child row: a foreign key constraint fails")
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: select distinct cola, colb from parent where (cola, colb) in ::__vals {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}} values:{type:TUPLE values:{type:INT64 value:"2"} values:{type:VARCHAR value:"b"}}} false false`,
+		})
+
+		vc.Rewind()
+		err = fkc.TryStreamExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true, func(result *sqltypes.Result) error { return nil })
+		require.ErrorContains(t, err, "Cannot add or update a child row: a foreign key constraint fails")
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`StreamExecuteMulti select distinct cola, colb from parent where (cola, colb) in ::__vals ks.0: {__vals: type:TUPLE values:{type:TUPLE values:{type:INT64 value:"1"} values:{type:VARCHAR value:"a"}} values:{type:TUPLE values:{type:INT64 value:"2"} values:{type:VARCHAR value:"b"}}} `,
+		})
+	})
+}