fix(replace): apply parent-side foreign key actions during REPLACE

pkg/sql/colexec/lockop/lock_op.go

@@ -248,7 +248,7 @@ func performLock(
 			target.primaryColumnType,
 			target.partitionColumnIndexInBatch,
 			DefaultLockOptions(lockOp.ctr.parker).
-				WithLockMode(lock.LockMode_Exclusive).
+				WithLockMode(target.mode).
 				WithFetchLockRowsFunc(lockOp.ctr.fetchers[idx]).
 				WithMaxBytesPerLock(int(proc.GetLockService().GetConfig().MaxLockRowCount)).
 				WithFilterRows(target.filter, filterCols).

pkg/sql/compile/compile.go

@@ -524,12 +524,15 @@ func (c *Compile) runOnce() (err error) {
 		c.proc.Base.StageCache.Clear()
 	}()
 
-	// Pre-check: REPLACE parent→child FK RESTRICT constraints must be
-	// verified before the REPLACE execution modifies any rows.
+	// REPLACE parent checks and actions run before the main pipeline.
 	query := c.pn.GetQuery()
 	if query != nil && query.StmtType == plan.Query_INSERT && len(query.GetDetectSqls()) != 0 {
 		for _, sql := range query.DetectSqls {
-			if strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") {
+			if strings.HasPrefix(sql, "REPLACE_PARENT_LOCK:") {
+				if err = c.runSql(strings.TrimPrefix(sql, "REPLACE_PARENT_LOCK:")); err != nil {
+					return err
+				}
+			} else if strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") {
 				if err = runDetectSql(c, strings.TrimPrefix(sql, "REPLACE_PARENT_CHK:")); err != nil {
 					// Only translate the "check returned false" signal into the
 					// parent-row-referenced error; pass through real execution
@@ -540,6 +543,10 @@ func (c *Compile) runOnce() (err error) {
 					}
 					return err
 				}
+			} else if strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:") {
+				if err = c.runSql(strings.TrimPrefix(sql, "REPLACE_PARENT_ACTION:")); err != nil {
+					return err
+				}
 			}
 		}
 	}
@@ -633,12 +640,13 @@ func (c *Compile) runOnce() (err error) {
 	query = c.pn.GetQuery()
 	if query != nil && (query.StmtType == plan.Query_INSERT ||
 		query.StmtType == plan.Query_UPDATE) && len(query.GetDetectSqls()) != 0 {
-		// Filter out pre-check SQLs (already executed before the main operation).
-		// The modern INSERT path enforces child→parent existence in-plan now, so the
-		// remaining DetectSqls are self-referencing FK checks (plain 1452 message).
+		// Filter out REPLACE parent-side checks and actions already executed before
+		// the main operation.
 		var postCheckSqls []string
 		for _, sql := range query.DetectSqls {
-			if strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") {
+			if strings.HasPrefix(sql, "REPLACE_PARENT_LOCK:") ||
+				strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") ||
+				strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:") {
 				continue
 			}
 			postCheckSqls = append(postCheckSqls, sql)

pkg/sql/compile/operator.go

@@ -786,7 +786,7 @@ func constructLockOp(node *plan.Node, eng engine.Engine) (*lockop.LockOp, error)
 			partitionColPos = target.PartitionColIdxInBat
 		}
 		typ := plan2.MakeTypeByPlan2Type(target.PrimaryColTyp)
-		arg.AddLockTarget(target.GetTableId(), target.GetObjRef(), target.GetPrimaryColIdxInBat(), typ, partitionColPos, target.GetRefreshTsIdxInBat(), target.GetLockRows(), target.GetLockTableAtTheEnd())
+		arg.AddLockTargetWithMode(target.GetTableId(), target.GetObjRef(), target.GetMode(), target.GetPrimaryColIdxInBat(), typ, partitionColPos, target.GetRefreshTsIdxInBat(), target.GetLockRows(), target.GetLockTableAtTheEnd())
 	}
 	for _, target := range node.LockTargets {
 		if target.LockTable {

pkg/sql/compile/remoterun.go

@@ -922,7 +922,7 @@ func convertToVmOperator(opr *pipeline.Instruction, ctx *scopeContext, eng engin
 		lockArg := lockop.NewArgumentByEngine(eng)
 		for _, target := range t.Targets {
 			typ := plan2.MakeTypeByPlan2Type(target.PrimaryColTyp)
-			lockArg.AddLockTarget(target.GetTableId(), target.GetObjRef(), target.GetPrimaryColIdxInBat(), typ, target.PartitionColIdxInBat, target.GetRefreshTsIdxInBat(), target.GetLockRows(), target.GetLockTableAtTheEnd())
+			lockArg.AddLockTargetWithMode(target.GetTableId(), target.GetObjRef(), target.GetMode(), target.GetPrimaryColIdxInBat(), typ, target.PartitionColIdxInBat, target.GetRefreshTsIdxInBat(), target.GetLockRows(), target.GetLockTableAtTheEnd())
 		}
 		for _, target := range t.Targets {
 			if target.LockTable {

pkg/sql/plan/bind_insert.go

@@ -24,6 +24,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/defines"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
+	lockpb "github.com/matrixorigin/matrixone/pkg/pb/lock"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 	"github.com/matrixorigin/matrixone/pkg/sql/util"
@@ -1062,6 +1063,25 @@ func (builder *QueryBuilder) appendModernChildFkMarkOks(
 			BindingTags:  []int32{parentTag},
 			ScanSnapshot: bindCtx.snapshot,
 		}, bindCtx)
+		pkPos, pkTyp := getPkPos(parentTableDef, false)
+		if pkPos < 0 {
+			return 0, nil, moerr.NewInternalErrorf(builder.GetContext(),
+				"foreign-key parent table %s has no lockable primary key", parentTableDef.Name)
+		}
+		parentScanID = builder.appendNode(&plan.Node{
+			NodeType: plan.Node_LOCK_OP,
+			Children: []int32{parentScanID},
+			TableDef: parentTableDef,
+			LockTargets: []*plan.LockTarget{{
+				TableId:            parentTableDef.TblId,
+				ObjRef:             parentObjRef,
+				PrimaryColIdxInBat: int32(pkPos),
+				PrimaryColRelPos:   parentTag,
+				PrimaryColTyp:      pkTyp,
+				Mode:               lockpb.LockMode_Shared,
+			}},
+			BindingTags: []int32{builder.genNewBindTag()},
+		}, bindCtx)
 
 		parentColId2Pos := make(map[uint64]int, len(parentTableDef.Cols))
 		for i, col := range parentTableDef.Cols {

pkg/sql/plan/bind_replace.go

@@ -242,11 +242,14 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 			})
 		}
 
-		var err error
 		for i, idxDef := range tableDef.Indexes {
 			if skipUniqueIdx[i] {
 				continue
 			}
+			prefixLengths, err := catalog.IndexPrefixLengthsFromParamsWithError(idxDef.IndexAlgoParams)
+			if err != nil {
+				return 0, err
+			}
 			idxObjRefs[i], idxTableDefs[i], err = builder.compCtx.ResolveIndexTableByRef(objRef, idxDef.IndexTableName, bindCtx.snapshot)
 			if err != nil {
 				return 0, err
@@ -255,11 +258,31 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 			oldColName2Idx[idxDef.IndexTableName+"."+catalog.IndexTablePrimaryColName] = oldColName2Idx[tableDef.Name+"."+tableDef.Pkey.PkeyColName]
 
 			if !indexTableStoresSerializedKey(idxDef) {
-				oldColName2Idx[idxDef.IndexTableName+"."+catalog.IndexTableIndexColName] = oldColName2Idx[tableDef.Name+"."+indexPrimaryPartName(idxDef)]
+				partName := indexPrimaryPartName(idxDef)
+				if prefixLengths[partName] > 0 {
+					colIdx := tableDef.Name2ColIndex[partName]
+					partExpr := &plan.Expr{
+						Typ: tableDef.Cols[colIdx].Typ,
+						Expr: &plan.Expr_Col{
+							Col: &plan.ColRef{RelPos: oldScanTag, ColPos: colIdx},
+						},
+					}
+					idxExpr, err := builder.makeIndexPartExprFromInputExpr(partExpr, partName, prefixLengths)
+					if err != nil {
+						return 0, err
+					}
+					oldColName2Idx[idxDef.IndexTableName+"."+catalog.IndexTableIndexColName] = [2]int32{
+						fullProjTag, int32(len(fullProjList)),
+					}
+					fullProjList = append(fullProjList, idxExpr)
+				} else {
+					oldColName2Idx[idxDef.IndexTableName+"."+catalog.IndexTableIndexColName] = oldColName2Idx[tableDef.Name+"."+partName]
+				}
 			} else {
 				args := make([]*plan.Expr, len(idxDef.Parts))
 				for j, part := range idxDef.Parts {
-					colIdx := tableDef.Name2ColIndex[catalog.ResolveAlias(part)]
+					partName := catalog.ResolveAlias(part)
+					colIdx := tableDef.Name2ColIndex[partName]
 					args[j] = &plan.Expr{
 						Typ: tableDef.Cols[colIdx].Typ,
 						Expr: &plan.Expr_Col{
@@ -269,6 +292,12 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 							},
 						},
 					}
+					if prefixLengths[partName] > 0 {
+						args[j], err = builder.makeIndexPartExprFromInputExpr(args[j], partName, prefixLengths)
+						if err != nil {
+							return 0, err
+						}
+					}
 				}
 
 				idxExpr := args[0]
@@ -300,6 +329,10 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 				if !idxDef.Unique || skipUniqueIdx[i] {
 					continue
 				}
+				prefixLengths, err := catalog.IndexPrefixLengthsFromParamsWithError(idxDef.IndexAlgoParams)
+				if err != nil {
+					return 0, err
+				}
 				var ukPartConds []*plan.Expr
 				for _, part := range idxDef.Parts {
 					colName := catalog.ResolveAlias(part)
@@ -323,6 +356,16 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 							},
 						},
 					}
+					if prefixLengths[colName] > 0 {
+						lExpr, err = builder.makeIndexPartExprFromInputExpr(lExpr, colName, prefixLengths)
+						if err != nil {
+							return 0, err
+						}
+						rExpr, err = builder.makeIndexPartExprFromInputExpr(rExpr, colName, prefixLengths)
+						if err != nil {
+							return 0, err
+						}
+					}
 					partCond, _ := BindFuncExprImplByPlanExpr(builder.GetContext(), "=", []*plan.Expr{lExpr, rExpr})
 					ukPartConds = append(ukPartConds, partCond)
 				}
@@ -363,6 +406,10 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 				if !idxDef.Unique || skipUniqueIdx[i] {
 					continue
 				}
+				prefixLengths, err := catalog.IndexPrefixLengthsFromParamsWithError(idxDef.IndexAlgoParams)
+				if err != nil {
+					return 0, err
+				}
 				var ukPartConds []*plan.Expr
 				for _, part := range idxDef.Parts {
 					colName := catalog.ResolveAlias(part)
@@ -386,6 +433,16 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 							},
 						},
 					}
+					if prefixLengths[colName] > 0 {
+						lExpr, err = builder.makeIndexPartExprFromInputExpr(lExpr, colName, prefixLengths)
+						if err != nil {
+							return 0, err
+						}
+						rExpr, err = builder.makeIndexPartExprFromInputExpr(rExpr, colName, prefixLengths)
+						if err != nil {
+							return 0, err
+						}
+					}
 					partCond, _ := BindFuncExprImplByPlanExpr(builder.GetContext(), "=", []*plan.Expr{lExpr, rExpr})
 					ukPartConds = append(ukPartConds, partCond)
 				}
@@ -843,7 +900,12 @@ func (builder *QueryBuilder) appendDedupAndMultiUpdateNodesForBindReplace(
 		deleteCols := make([]plan.ColRef, 2)
 
 		newIdxPos := colName2Idx[idxDef.IndexTableName+"."+catalog.IndexTableIndexColName]
-		if indexTableStoresSerializedKey(idxDef) {
+		prefixLengths, err := catalog.IndexPrefixLengthsFromParamsWithError(idxDef.IndexAlgoParams)
+		if err != nil {
+			return 0, err
+		}
+		partName := indexPrimaryPartName(idxDef)
+		if indexTableStoresSerializedKey(idxDef) || prefixLengths[partName] > 0 {
 			idxExpr := &plan.Expr{
 				Typ: fullProjList[newIdxPos].Typ,
 				Expr: &plan.Expr_Col{
@@ -1136,8 +1198,23 @@ func (builder *QueryBuilder) appendNodesForReplaceStmt(
 
 		idxTableName := idxDef.IndexTableName
 		colName2Idx[idxTableName+"."+catalog.IndexTablePrimaryColName] = pkPos
+		prefixLengths, err := catalog.IndexPrefixLengthsFromParamsWithError(idxDef.IndexAlgoParams)
+		if err != nil {
+			return 0, nil, nil, err
+		}
 		if !indexTableStoresSerializedKey(idxDef) {
-			colName2Idx[idxTableName+"."+catalog.IndexTableIndexColName] = colName2Idx[tableDef.Name+"."+indexPrimaryPartName(idxDef)]
+			partName := indexPrimaryPartName(idxDef)
+			partPos := colName2Idx[tableDef.Name+"."+partName]
+			if prefixLengths[partName] > 0 {
+				idxExpr, err := builder.makeIndexPartExprFromInputExpr(projList2[partPos], partName, prefixLengths)
+				if err != nil {
+					return 0, nil, nil, err
+				}
+				colName2Idx[idxTableName+"."+catalog.IndexTableIndexColName] = int32(len(projList2))
+				projList2 = append(projList2, idxExpr)
+			} else {
+				colName2Idx[idxTableName+"."+catalog.IndexTableIndexColName] = partPos
+			}
 		} else {
 			argsLen := len(idxDef.Parts)
 			args := make([]*plan.Expr, argsLen)
@@ -1149,7 +1226,15 @@ func (builder *QueryBuilder) appendNodesForReplaceStmt(
 					errMsg := fmt.Sprintf("bind insert err, can not find colName = %s", idxDef.Parts[k])
 					return 0, nil, nil, moerr.NewInternalError(builder.GetContext(), errMsg)
 				}
-				args[k] = DeepCopyExpr(projList2[colPos])
+				partName := catalog.ResolveAlias(idxDef.Parts[k])
+				if prefixLengths[partName] > 0 {
+					args[k], err = builder.makeIndexPartExprFromInputExpr(projList2[colPos], partName, prefixLengths)
+					if err != nil {
+						return 0, nil, nil, err
+					}
+				} else {
+					args[k] = DeepCopyExpr(projList2[colPos])
+				}
 			}
 
 			funcName := "serial"

pkg/sql/plan/build.go

@@ -178,7 +178,29 @@ func bindAndOptimizeReplaceQuery(ctx CompilerContext, stmt *tree.Replace, isPrep
 	if err != nil {
 		return nil, err
 	}
-	if len(tblInfo.tableDefs) == 1 {
+	// FK checks/actions are all disabled when foreign_key_checks is off, the
+	// same way MySQL skips foreign-key enforcement. Gate every FK SQL below
+	// (self-referencing checks, the RESTRICT pre-check, and the non-self
+	// parent-side actions) under one guard so the behavior is consistent.
+	fkChecksEnabled, err := IsForeignKeyChecksEnabled(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if fkChecksEnabled && len(tblInfo.tableDefs) == 1 {
+		parentLock, parentChecks, parentActions, err := genParentSideReplaceFKSqls(
+			ctx, tblInfo.objRef[0], tblInfo.tableDefs[0], stmt)
+		if err != nil {
+			return nil, err
+		}
+		if parentLock != "" {
+			query.DetectSqls = append(query.DetectSqls, "REPLACE_PARENT_LOCK:"+parentLock)
+		}
+		for _, sql := range parentChecks {
+			query.DetectSqls = append(query.DetectSqls, "REPLACE_PARENT_CHK:"+sql)
+		}
+		for _, sql := range parentActions {
+			query.DetectSqls = append(query.DetectSqls, "REPLACE_PARENT_ACTION:"+sql)
+		}
 		sqls, err := genSqlsForCheckFKSelfRefer(
 			ctx.GetContext(),
 			tblInfo.objRef[0].SchemaName,
@@ -189,7 +211,7 @@ func bindAndOptimizeReplaceQuery(ctx CompilerContext, stmt *tree.Replace, isPrep
 		if err != nil {
 			return nil, err
 		}
-		query.DetectSqls = sqls
+		query.DetectSqls = append(query.DetectSqls, sqls...)
 
 		// Generate pre-check SQLs for parent→child safety (RESTRICT).
 		preCheckSqls, err := genPreCheckSqlsForReplaceFKSelfRefer(
@@ -206,6 +228,7 @@ func bindAndOptimizeReplaceQuery(ctx CompilerContext, stmt *tree.Replace, isPrep
 		for _, sql := range preCheckSqls {
 			query.DetectSqls = append(query.DetectSqls, "REPLACE_PARENT_CHK:"+sql)
 		}
+
 	}
 
 	return &Plan{