Disallow Insert with Duplicate key update and Replace Into queries on foreign key column, set locks on fk queries

go/vt/sqlparser/ast_funcs.go

@@ -2533,13 +2533,22 @@ func (v *visitor) visitAllSelects(in SelectStatement, f func(p *Select, idx int)
 	panic("switch should be exhaustive")
 }
 
-func IsNonLiteral(updExprs UpdateExprs) bool {
-	for _, updateExpr := range updExprs {
-		switch updateExpr.Expr.(type) {
-		case *Argument, *NullVal, BoolVal, *Literal:
-		default:
-			return true
-		}
+// IsRestrict returns true if the reference action is of restrict type.
+func (ra ReferenceAction) IsRestrict() bool {
+	switch ra {
+	case Restrict, NoAction, DefaultAction:
+		return true
+	default:
+		return false
+	}
+}
+
+// IsLiteral returns true if the expression is of a literal type.
+func IsLiteral(expr Expr) bool {
+	switch expr.(type) {
+	case *Argument, *NullVal, BoolVal, *Literal:
+		return true
+	default:
+		return false
 	}
-	return false
 }

go/vt/vtgate/planbuilder/insert.go

@@ -18,6 +18,7 @@ package planbuilder
 
 import (
 	querypb "vitess.io/vitess/go/vt/proto/query"
+	vschemapb "vitess.io/vitess/go/vt/proto/vschema"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vtgate/engine"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/operators"
@@ -44,9 +45,11 @@ func gen4InsertStmtPlanner(version querypb.ExecuteOptions_PlannerVersion, insStm
 	// We cannot shortcut here as sequence column needs additional planning.
 	ks, tables := ctx.SemTable.SingleUnshardedKeyspace()
 	if ks != nil && tables[0].AutoIncrement == nil {
-		plan := insertUnshardedShortcut(insStmt, ks, tables)
-		plan = pushCommentDirectivesOnPlan(plan, insStmt)
-		return newPlanResult(plan.Primitive(), operators.QualifiedTables(ks, tables)...), nil
+		if fkManagementNotRequiredForInsert(ctx, tables[0], sqlparser.UpdateExprs(insStmt.OnDup)) {
+			plan := insertUnshardedShortcut(insStmt, ks, tables)
+			plan = pushCommentDirectivesOnPlan(plan, insStmt)
+			return newPlanResult(plan.Primitive(), operators.QualifiedTables(ks, tables)...), nil
+		}
 	}
 
 	tblInfo, err := ctx.SemTable.TableInfoFor(ctx.SemTable.TableSetFor(insStmt.Table))
@@ -83,6 +86,26 @@ func gen4InsertStmtPlanner(version querypb.ExecuteOptions_PlannerVersion, insStm
 	return newPlanResult(plan.Primitive(), operators.TablesUsed(op)...), nil
 }
 
+// TODO: Handle all this in semantic analysis.
+func fkManagementNotRequiredForInsert(ctx *plancontext.PlanningContext, vTbl *vindexes.Table, updateExprs sqlparser.UpdateExprs) bool {
+	ksMode, err := ctx.VSchema.ForeignKeyMode(vTbl.Keyspace.Name)
+	if err != nil || ksMode != vschemapb.Keyspace_FK_MANAGED {
+		return true
+	}
+
+	if len(vTbl.ParentFKsNeedsHandling(ctx.VerifyAllFKs, "")) > 0 {
+		return false
+	}
+
+	childFks := vTbl.ChildFKsNeedsHandling(ctx.VerifyAllFKs, vindexes.UpdateAction)
+	getFKInfo := func(expr *sqlparser.UpdateExpr) ([]vindexes.ParentFKInfo, []vindexes.ChildFKInfo) {
+		return nil, childFks
+	}
+
+	// Check if any column in the parent table is being updated which has a child foreign key.
+	return !columnModified(updateExprs, getFKInfo)
+}
+
 func insertUnshardedShortcut(stmt *sqlparser.Insert, ks *vindexes.Keyspace, tables []*vindexes.Table) logicalPlan {
 	eIns := &engine.Insert{}
 	eIns.Keyspace = ks