remove fk_verify operator, refactors the fk test

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

go/vt/vtgate/planbuilder/insert.go

@@ -101,9 +101,8 @@ func insertUnshardedShortcut(stmt *sqlparser.Insert, ks *vindexes.Keyspace, tabl
 }
 
 type insert struct {
-	eInsert  *engine.Insert
-	source   logicalPlan
-	fkVerify logicalPlan
+	eInsert *engine.Insert
+	source  logicalPlan
 }
 
 var _ logicalPlan = (*insert)(nil)

go/vt/vtgate/planbuilder/operator_transformers.go

@@ -66,8 +66,6 @@ func transformToLogicalPlan(ctx *plancontext.PlanningContext, op ops.Operator, i
 		return transformAggregator(ctx, op)
 	case *operators.Distinct:
 		return transformDistinct(ctx, op)
-	case *operators.FKVerify:
-		return nil, vterrors.VT12002()
 	}
 
 	return nil, vterrors.VT13001(fmt.Sprintf("unknown type encountered: %T (transformToLogicalPlan)", op))
@@ -429,13 +427,6 @@ func transformInsertPlan(ctx *plancontext.PlanningContext, op *operators.Route,
 		}
 	}
 
-	if ins.FKVerify != nil {
-		i.fkVerify, err = transformToLogicalPlan(ctx, ins.FKVerify, true)
-		if err != nil {
-			return
-		}
-	}
-
 	return
 }
 

go/vt/vtgate/planbuilder/operators/ast2op.go

@@ -269,10 +269,7 @@ func createOperatorFromInsert(ctx *plancontext.PlanningContext, ins *sqlparser.I
 	if ksMode == vschemapb.Keyspace_FK_MANAGED {
 		parentFKs := vindexTable.CrossShardParentFKs()
 		if len(parentFKs) > 0 {
-			insOp.FKVerify, err = NewFkVerify(ctx, parentFKs, ins.Columns)
-			if err != nil {
-				return nil, err
-			}
+			return nil, vterrors.VT12002()
 		}
 	}
 

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

@@ -50,10 +50,6 @@ type Insert struct {
 	// Insert using select query will have select plan as input operator for the insert operation.
 	Input ops.Operator
 
-	// FKVerify is required to check the validity of foreign key constraints that aren't shard scoped.
-	// It is nil if no verification is required.
-	FKVerify ops.Operator
-
 	noColumns
 	noPredicates
 }
@@ -115,7 +111,6 @@ func (i *Insert) Clone(inputs []ops.Operator) ops.Operator {
 		ColVindexes:       i.ColVindexes,
 		VindexValues:      i.VindexValues,
 		VindexValueOffset: i.VindexValueOffset,
-		FKVerify:          i.FKVerify,
 	}
 }
 

go/vt/vtgate/vindexes/foreign_keys.go

@@ -112,7 +112,6 @@ func (t *Table) CrossShardParentFKs() (crossShardFKs []ParentFKInfo) {
 	if len(t.ParentForeignKeys) == 0 {
 		return
 	}
-	primaryVindex := t.ColumnVindexes[0]
 	for _, fk := range t.ParentForeignKeys {
 		// If the keyspaces are different, then the fk definition
 		// is going to go across shards.
@@ -125,8 +124,10 @@ func (t *Table) CrossShardParentFKs() (crossShardFKs []ParentFKInfo) {
 		if !t.Keyspace.Sharded {
 			continue
 		}
+
 		// If the primary vindexes don't match between the parent and child table,
 		// we cannot infer that the fk constraint in shard scoped.
+		primaryVindex := t.ColumnVindexes[0]
 		if fk.Table.ColumnVindexes[0].Vindex != primaryVindex.Vindex {
 			crossShardFKs = append(crossShardFKs, fk)
 			continue