fix(plan): move parent FK update actions to MULTI_UPDATE

pkg/sql/colexec/multi_update/affect_rows_test.go

@@ -78,6 +78,14 @@ func TestUpsertAffectRowsAccounting(t *testing.T) {
 		require.EqualValues(t, 0, op.GetAffectedRows())
 	})
 
+	t.Run("foreign key side effects never affect rows", func(t *testing.T) {
+		op := newAffectRowsTestOp(actionUpdate, true)
+		op.MultiUpdateCtx = []*MultiUpdateCtx{{IgnoreAffectedRows: true}}
+		op.addInsertAffectRows(UpdateMainTable, 4)
+		op.addDeleteAffectRows(UpdateMainTable, 3)
+		require.EqualValues(t, 0, op.GetAffectedRows())
+	})
+
 	t.Run("batch upsert: new + updated rows", func(t *testing.T) {
 		// 2 brand new rows (INSERT only) + 3 conflicting rows (DELETE + INSERT):
 		// inserts cover all 5 rows, deletes cover the 3 conflicts => 5 + 3 = 8.

pkg/sql/colexec/multi_update/multi_update_partition.go

@@ -443,6 +443,7 @@ func (ctx *MultiUpdateCtx) clone() *MultiUpdateCtx {
 		PartitionCols:      ctx.PartitionCols,
 		SkipInsertOnNullPk: ctx.SkipInsertOnNullPk,
 		InsertPkColIdx:     ctx.InsertPkColIdx,
+		IgnoreAffectedRows: ctx.IgnoreAffectedRows,
 	}
 	objRef := *ctx.ObjRef
 	def := *ctx.TableDef

pkg/sql/colexec/multi_update/types.go

@@ -116,7 +116,8 @@ type MultiUpdateCtx struct {
 	SkipInsertOnNullPk bool
 	// InsertPkColIdx is the PK column's index within InsertCols. It is only
 	// used with SkipInsertOnNullPk for REPLACE delete-only rows.
-	InsertPkColIdx int
+	InsertPkColIdx     int
+	IgnoreAffectedRows bool
 }
 
 func (update MultiUpdate) TypeName() string {
@@ -230,5 +231,8 @@ func (update *MultiUpdate) addDeleteAffectRows(tableType UpdateTableType, rowCou
 }
 
 func (update *MultiUpdate) doAddAffectedRows(affectedRows uint64) {
+	if len(update.MultiUpdateCtx) > 0 && update.MultiUpdateCtx[0].IgnoreAffectedRows {
+		return
+	}
 	update.ctr.affectedRows += affectedRows
 }

pkg/sql/compile/operator.go

@@ -896,6 +896,7 @@ func constructMultiUpdate(
 			PartitionCols:      partitionCols,
 			SkipInsertOnNullPk: updateCtx.SkipInsertOnNullPk,
 			InsertPkColIdx:     int(updateCtx.InsertPkColIdx),
+			IgnoreAffectedRows: updateCtx.IgnoreAffectedRows,
 		}
 	}
 	arg.Action = action

pkg/sql/compile/remoterun.go

@@ -867,6 +867,7 @@ func convertToPipelineInstruction(op vm.Operator, proc *process.Process, ctx *sc
 				TableDef:              muCtx.TableDef,
 				SkipInsertOnNullPk:    muCtx.SkipInsertOnNullPk,
 				InsertPkColIdx:        int32(muCtx.InsertPkColIdx),
+				IgnoreAffectedRows:    muCtx.IgnoreAffectedRows,
 				CountDeleteAffectRows: t.CountDeleteAffectRows,
 			}
 
@@ -1374,6 +1375,7 @@ func convertToVmOperator(opr *pipeline.Instruction, ctx *scopeContext, eng engin
 				TableDef:           muCtx.TableDef,
 				SkipInsertOnNullPk: muCtx.SkipInsertOnNullPk,
 				InsertPkColIdx:     int(muCtx.InsertPkColIdx),
+				IgnoreAffectedRows: muCtx.IgnoreAffectedRows,
 			}
 
 			arg.MultiUpdateCtx[i].InsertCols = make([]int, len(muCtx.InsertCols))

pkg/sql/compile/remoterun_test.go

@@ -716,8 +716,9 @@ func Test_DMLOperatorSerializationRoundtrip(t *testing.T) {
 		op := &multi_update.MultiUpdate{
 			MultiUpdateCtx: []*multi_update.MultiUpdateCtx{
 				{
-					ObjRef:   &plan.ObjectRef{ObjName: "t1"},
-					TableDef: &plan.TableDef{Name: "t1"},
+					ObjRef:             &plan.ObjectRef{ObjName: "t1"},
+					TableDef:           &plan.TableDef{Name: "t1"},
+					IgnoreAffectedRows: true,
 				},
 			},
 			Action:                multi_update.UpdateWriteTable,
@@ -727,12 +728,16 @@ func Test_DMLOperatorSerializationRoundtrip(t *testing.T) {
 		require.NoError(t, err)
 		require.True(t, pipeInstr.MultiUpdate.UpdateCtxList[0].CountDeleteAffectRows,
 			"serialized UpdateCtx must carry CountDeleteAffectRows")
+		require.True(t, pipeInstr.MultiUpdate.UpdateCtxList[0].IgnoreAffectedRows,
+			"serialized UpdateCtx must carry IgnoreAffectedRows")
 
 		restored, err := convertToVmOperator(pipeInstr, ctx, nil)
 		require.NoError(t, err)
 		restoredOp := restored.(*multi_update.MultiUpdate)
 		require.True(t, restoredOp.CountDeleteAffectRows,
 			"CountDeleteAffectRows must survive the remote pipeline round-trip")
+		require.True(t, restoredOp.MultiUpdateCtx[0].IgnoreAffectedRows,
+			"IgnoreAffectedRows must survive the remote pipeline round-trip")
 	})
 
 	t.Run("MultiUpdate_RejectZeroTemporal", func(t *testing.T) {

pkg/sql/plan/bind_insert.go

@@ -1061,7 +1061,7 @@ func (builder *QueryBuilder) appendModernChildFkMarkOks(
 	selectTag int32,
 	childColPos func(colName string) int32,
 ) (int32, []*plan.Expr, error) {
-	selectNode := builder.qry.Nodes[lastNodeID]
+	selectNode := builder.updateInputProjectNode(lastNodeID)
 	inputTypes := make([]plan.Type, len(selectNode.ProjectList))
 	for i, expr := range selectNode.ProjectList {
 		inputTypes[i] = expr.Typ

pkg/sql/plan/bind_update.go

@@ -85,6 +85,17 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext)
 	if err != nil {
 		return 0, err
 	}
+	for i, tableDef := range dmlCtx.tableDefs {
+		if len(dmlCtx.updateCol2Expr[i]) > 0 &&
+			(len(tableDef.Fkeys) > 0 || len(tableDef.RefChildTbls) > 0) {
+			// FK validation and parent-side actions depend on the current
+			// foreign_key_checks value. Mark the plan before that value is
+			// inspected so a plan built with checks disabled cannot be reused
+			// after they are enabled (or vice versa).
+			builder.qry.HasForeignKeyAction = true
+			break
+		}
+	}
 	if err = validateUpdateTargetSubqueries(builder.compCtx, stmt, dmlCtx.objRefs, dmlCtx.tableDefs); err != nil {
 		return 0, err
 	}
@@ -399,44 +410,6 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext)
 		}
 	}
 
-	fkChecksEnabled, err := IsForeignKeyChecksEnabled(builder.compCtx)
-	if err != nil {
-		return 0, err
-	}
-	if updateMayDependOnForeignKeys(dmlCtx, newColName2Idx) {
-		// The plan shape and planner route depend on foreign_key_checks.
-		// Preserve that dependency even while checks are disabled so prepared
-		// and generic plan caches rebuild after either session-state transition.
-		builder.qry.HasForeignKeyAction = true
-	}
-	if fkChecksEnabled {
-		for i, tableDef := range dmlCtx.tableDefs {
-			if updateAutoIncrCols[i] &&
-				len(affectedUpdateChildFks(tableDef, dmlCtx.aliases[i], newColName2Idx)) > 0 {
-				return 0, newLegacyUpdatePlannerRouteError(
-					updateRouteReasonAutoIncrement,
-					moerr.NewUnsupportedDML(
-						builder.compCtx.GetContext(),
-						"auto_increment foreign key update",
-					),
-				)
-			}
-		}
-	}
-
-	lastNodeID, selectNodeTag, selectNode, err = builder.appendUpdateForeignKeyChecks(
-		bindCtx,
-		dmlCtx,
-		lastNodeID,
-		selectNodeTag,
-		oldColName2Idx,
-		newColName2Idx,
-		fkChecksEnabled,
-	)
-	if err != nil {
-		return 0, err
-	}
-
 	for i, tableDef := range dmlCtx.tableDefs {
 		if updateAutoIncrCols[i] {
 			lastNodeID = builder.appendNode(&plan.Node{
@@ -453,6 +426,19 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext)
 		}
 	}
 
+	lastNodeID, selectNodeTag, selectNode, err = builder.appendUpdateForeignKeyChecks(
+		bindCtx,
+		dmlCtx,
+		lastNodeID,
+		selectNodeTag,
+		selectNode,
+		oldColName2Idx,
+		newColName2Idx,
+	)
+	if err != nil {
+		return 0, err
+	}
+
 	idxScanNodes := make([][]*plan.Node, len(dmlCtx.tableDefs))
 	pkNeedUpdate := make([]bool, len(dmlCtx.tableDefs))
 	idxNeedUpdate := make([][]bool, len(dmlCtx.tableDefs))