return verify error based on type

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

go/vt/vtgate/engine/fk_verify.go

@@ -40,6 +40,11 @@ type FkVerify struct {
 	txNeeded
 }
 
+const (
+	ParentVerify = "VerifyParent"
+	ChildVerify  = "VerifyChild"
+)
+
 // RouteType implements the Primitive interface
 func (f *FkVerify) RouteType() string {
 	return "FKVerify"
@@ -68,7 +73,7 @@ func (f *FkVerify) TryExecute(ctx context.Context, vcursor VCursor, bindVars map
 			return nil, err
 		}
 		if len(qr.Rows) > 0 {
-			return nil, vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.NoReferencedRow2, "Cannot add or update a child row: a foreign key constraint fails")
+			return nil, getError(v.Typ)
 		}
 	}
 	return vcursor.ExecutePrimitive(ctx, f.Exec, bindVars, wantfields)
@@ -79,7 +84,7 @@ func (f *FkVerify) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVa
 	for _, v := range f.Verify {
 		err := vcursor.StreamExecutePrimitive(ctx, v.Exec, bindVars, wantfields, func(qr *sqltypes.Result) error {
 			if len(qr.Rows) > 0 {
-				return vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.NoReferencedRow2, "Cannot add or update a child row: a foreign key constraint fails")
+				return getError(v.Typ)
 			}
 			return nil
 		})
@@ -113,3 +118,10 @@ func (f *FkVerify) description() PrimitiveDescription {
 }
 
 var _ Primitive = (*FkVerify)(nil)
+
+func getError(typ string) error {
+	if typ == ParentVerify {
+		return vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.NoReferencedRow2, "Cannot add or update a child row: a foreign key constraint fails")
+	}
+	return vterrors.NewErrorf(vtrpcpb.Code_FAILED_PRECONDITION, vterrors.RowIsReferenced2, "Cannot delete or update a parent row: a foreign key constraint fails")
+}

go/vt/vtgate/engine/fk_verify_test.go

@@ -35,6 +35,13 @@ func TestFKVerifyUpdate(t *testing.T) {
 			Keyspace: &vindexes.Keyspace{Name: "ks"},
 		},
 	}
+	verifyC := &Route{
+		Query: "select 1 from grandchild g join child c on g.cola = c.cola and g.colb = c.colb where c.foo = 48",
+		RoutingParameters: &RoutingParameters{
+			Opcode:   Unsharded,
+			Keyspace: &vindexes.Keyspace{Name: "ks"},
+		},
+	}
 	childP := &Update{
 		DML: &DML{
 			Query: "update child set cola = 1, colb = 'a' where foo = 48",
@@ -45,7 +52,7 @@ func TestFKVerifyUpdate(t *testing.T) {
 		},
 	}
 	fkc := &FkVerify{
-		Verify: []Primitive{verifyP},
+		Verify: []*Verify{{Exec: verifyP, Typ: ParentVerify}},
 		Exec:   childP,
 	}
 
@@ -73,7 +80,7 @@ func TestFKVerifyUpdate(t *testing.T) {
 		})
 	})
 
-	t.Run("foreign key verification failure", func(t *testing.T) {
+	t.Run("parent 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("1", "int64"), "1", "1", "1")
 		vc := newDMLTestVCursor("0")
@@ -93,4 +100,26 @@ func TestFKVerifyUpdate(t *testing.T) {
 			`StreamExecuteMulti select 1 from child c left join parent p on p.cola = 1 and p.colb = 'a' where p.cola is null and p.colb is null ks.0: {} `,
 		})
 	})
+
+	fkc.Verify[0] = &Verify{Exec: verifyC, Typ: ChildVerify}
+	t.Run("child 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("1", "int64"), "1", "1", "1")
+		vc := newDMLTestVCursor("0")
+		vc.results = []*sqltypes.Result{fakeRes}
+		_, err := fkc.TryExecute(context.Background(), vc, map[string]*querypb.BindVariable{}, true)
+		require.ErrorContains(t, err, "Cannot delete or update a parent row: a foreign key constraint fails")
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`ExecuteMultiShard ks.0: select 1 from grandchild g join child c on g.cola = c.cola and g.colb = c.colb where c.foo = 48 {} 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 delete or update a parent row: a foreign key constraint fails")
+		vc.ExpectLog(t, []string{
+			`ResolveDestinations ks [] Destinations:DestinationAllShards()`,
+			`StreamExecuteMulti select 1 from grandchild g join child c on g.cola = c.cola and g.colb = c.colb where c.foo = 48 ks.0: {} `,
+		})
+	})
 }

go/vt/vtgate/planbuilder/operators/ast_to_update_op.go

@@ -385,7 +385,7 @@ func createFKVerifyOp(ctx *plancontext.PlanningContext, childOp ops.Operator, up
 		}
 		Verify = append(Verify, &VerifyOp{
 			Op:  op,
-			Typ: ParentVerify,
+			Typ: engine.ParentVerify,
 		})
 	}
 	// This validates that the old values don't exist on the child table.
@@ -396,7 +396,7 @@ func createFKVerifyOp(ctx *plancontext.PlanningContext, childOp ops.Operator, up
 		}
 		Verify = append(Verify, &VerifyOp{
 			Op:  op,
-			Typ: ChildVerify,
+			Typ: engine.ChildVerify,
 		})
 	}
 

go/vt/vtgate/planbuilder/operators/fk_verify.go

@@ -27,11 +27,6 @@ type VerifyOp struct {
 	Typ string
 }
 
-const (
-	ParentVerify = "VerifyParent"
-	ChildVerify  = "VerifyChild"
-)
-
 // FkVerify is used to represent a foreign key verification operation
 // as an operator. This operator is created for DML queries that require
 // verifications on the existence of the rows in the parent table (for example, INSERT and UPDATE).