diff --git a/pkg/sql/colexec/lockop/lock_op.go b/pkg/sql/colexec/lockop/lock_op.go index 76ece2e2349dc..2c0f33023d55c 100644 --- a/pkg/sql/colexec/lockop/lock_op.go +++ b/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). @@ -313,6 +313,17 @@ func LockTable( tableID uint64, pkType types.Type, changeDef bool) error { + return LockTableWithMode(eng, proc, tableID, pkType, lock.LockMode_Exclusive, changeDef) +} + +// LockTableWithMode locks all rows in a table with the specified lock mode. +func LockTableWithMode( + eng engine.Engine, + proc *process.Process, + tableID uint64, + pkType types.Type, + mode lock.LockMode, + changeDef bool) error { txnOp := proc.GetTxnOperator() if !txnOp.Txn().IsPessimistic() { return nil @@ -336,6 +347,7 @@ func LockTable( }() opts := DefaultLockOptions(parker). + WithLockMode(mode). WithLockTable(true, changeDef). WithFetchLockRowsFunc(GetFetchRowsFunc(pkType)) _, defChanged, refreshTS, err := doLock( @@ -1540,7 +1552,8 @@ func lockTalbeIfLockCountIsZero( if !target.lockTableAtTheEnd { continue } - err := LockTable(lockOp.engine, proc, target.tableID, target.primaryColumnType, false) + err := LockTableWithMode( + lockOp.engine, proc, target.tableID, target.primaryColumnType, target.mode, false) if err != nil { return err } diff --git a/pkg/sql/colexec/lockop/lock_op_test.go b/pkg/sql/colexec/lockop/lock_op_test.go index f14a5827acff0..3009e49a5bd0b 100644 --- a/pkg/sql/colexec/lockop/lock_op_test.go +++ b/pkg/sql/colexec/lockop/lock_op_test.go @@ -1087,8 +1087,9 @@ func TestCallLockOpLocksTableAtEOFWhenNoRowsProduced(t *testing.T) { IsFirst: false, IsLast: false, } - arg.AddLockTarget(tableID, nil, 0, pkType, -1, -1, nil, true) - arg.LockTable(tableID, false) + arg.AddLockTargetWithMode( + tableID, nil, lock.LockMode_Shared, 0, pkType, -1, -1, nil, true) + arg.LockTableWithMode(tableID, lock.LockMode_Shared, false) resetChildren(arg, nil) defer arg.Free(proc, false, nil) @@ -1096,6 +1097,16 @@ func TestCallLockOpLocksTableAtEOFWhenNoRowsProduced(t *testing.T) { _, err := vm.Exec(arg, proc) require.NoError(t, err) require.True(t, proc.GetTxnOperator().HasLockTable(tableID)) + + sharedTxn, err := proc.Base.TxnClient.New(proc.Ctx, timestamp.Timestamp{}) + require.NoError(t, err) + defer func() { require.NoError(t, sharedTxn.Rollback(proc.Ctx)) }() + sharedProc := process.NewTopProcess(proc.Ctx, mpool.MustNewZero(), proc.Base.TxnClient, + sharedTxn, nil, proc.GetLockService(), nil, nil, nil, nil, nil) + require.NoError(t, LockTableWithMode( + nil, sharedProc, tableID, pkType, lock.LockMode_Shared, false)) + require.NoError(t, LockTable(nil, proc, tableID+1, pkType, false)) + require.True(t, proc.GetTxnOperator().HasLockTable(tableID+1)) }, ) } diff --git a/pkg/sql/compile/compile.go b/pkg/sql/compile/compile.go index 1efd70ac4fa98..94d5017d0813a 100644 --- a/pkg/sql/compile/compile.go +++ b/pkg/sql/compile/compile.go @@ -525,12 +525,19 @@ 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 { + if err = validateReplaceParentTxnMode( + c.proc.Ctx, query, c.proc.GetTxnOperator().Txn().IsPessimistic()); err != nil { + return err + } 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 @@ -541,6 +548,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 + } } } } @@ -634,12 +645,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) @@ -656,6 +668,19 @@ func (c *Compile) runOnce() (err error) { return err } +func validateReplaceParentTxnMode(ctx context.Context, query *plan.Query, pessimistic bool) error { + if pessimistic || query == nil { + return nil + } + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_LOCK:") { + return moerr.NewNotSupported(ctx, + "REPLACE on a referenced parent table in optimistic transaction mode") + } + } + return nil +} + // add log to check if background sql return NeedRetry error when origin sql execute successfully func (c *Compile) debugLogFor19288(err error, bsql string) { if c.isRetryErr(err) { @@ -807,11 +832,12 @@ func (c *Compile) lockTable() error { for _, tableID := range tableIDs { tbl := c.lockTables[tableID] typ := plan2.MakeTypeByPlan2Type(tbl.PrimaryColTyp) - if err := lockop.LockTable( + if err := lockop.LockTableWithMode( c.e, c.proc, tbl.TableId, typ, + tbl.Mode, false); err != nil { return err } diff --git a/pkg/sql/compile/compile_test.go b/pkg/sql/compile/compile_test.go index c3c8509f2fc6e..e834328529fdd 100644 --- a/pkg/sql/compile/compile_test.go +++ b/pkg/sql/compile/compile_test.go @@ -32,8 +32,11 @@ import ( "github.com/matrixorigin/matrixone/pkg/container/vector" "github.com/matrixorigin/matrixone/pkg/defines" "github.com/matrixorigin/matrixone/pkg/lockservice" + lockpb "github.com/matrixorigin/matrixone/pkg/pb/lock" + "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/pb/timestamp" "github.com/matrixorigin/matrixone/pkg/perfcounter" + "github.com/matrixorigin/matrixone/pkg/sql/colexec/lockop" "github.com/matrixorigin/matrixone/pkg/txn/client" "github.com/matrixorigin/matrixone/pkg/txn/rpc" @@ -45,7 +48,6 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/buffer" "github.com/matrixorigin/matrixone/pkg/container/batch" mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test" - "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/pb/txn" "github.com/matrixorigin/matrixone/pkg/sql/colexec" "github.com/matrixorigin/matrixone/pkg/sql/colexec/group" @@ -214,6 +216,36 @@ func TestShouldPrePipelineLockTable(t *testing.T) { require.False(t, target.LockTableAtTheEnd) } +func TestConstructLockOpPreservesSharedTableMode(t *testing.T) { + for _, lockTable := range []bool{false, true} { + t.Run(fmt.Sprintf("table=%t", lockTable), func(t *testing.T) { + node := &plan.Node{LockTargets: []*plan.LockTarget{{ + TableId: 42, PrimaryColTyp: plan.Type{Id: int32(types.T_int64)}, + Mode: lockpb.LockMode_Shared, LockTable: lockTable, + }}} + + op, err := constructLockOp(node, nil) + require.NoError(t, err) + targets := op.CopyToPipelineTarget() + require.Len(t, targets, 1) + assert.Equal(t, lockTable, targets[0].LockTable) + assert.Equal(t, lockpb.LockMode_Shared, targets[0].Mode) + }) + } +} + +func TestValidateReplaceParentTxnMode(t *testing.T) { + ctx := context.Background() + query := &plan.Query{DetectSqls: []string{"REPLACE_PARENT_LOCK:select 1 for update"}} + + require.NoError(t, validateReplaceParentTxnMode(ctx, query, true)) + require.ErrorContains(t, validateReplaceParentTxnMode(ctx, query, false), + "optimistic transaction mode") + require.NoError(t, validateReplaceParentTxnMode(ctx, + &plan.Query{DetectSqls: []string{"select true"}}, false)) + require.NoError(t, validateReplaceParentTxnMode(ctx, nil, false)) +} + func TestLockTableLocksAllPrePipelineTargets(t *testing.T) { runtime.RunTest( "", @@ -259,7 +291,8 @@ func TestLockTableLocksAllPrePipelineTargets(t *testing.T) { c := &Compile{ proc: proc, lockTables: map[uint64]*plan.LockTarget{ - 10: {TableId: 10, PrimaryColTyp: plan.Type{Id: int32(types.T_int32)}}, + 10: {TableId: 10, PrimaryColTyp: plan.Type{Id: int32(types.T_int32)}, + Mode: lockpb.LockMode_Shared}, 11: {TableId: 11, PrimaryColTyp: plan.Type{Id: int32(types.T_int32)}}, }, } @@ -267,6 +300,14 @@ func TestLockTableLocksAllPrePipelineTargets(t *testing.T) { require.NoError(t, c.lockTable()) require.True(t, txnOp.HasLockTable(10)) require.True(t, txnOp.HasLockTable(11)) + + sharedTxn, err := txnClient.New(ctx, timestamp.Timestamp{}) + require.NoError(t, err) + defer func() { require.NoError(t, sharedTxn.Rollback(ctx)) }() + sharedProc := process.NewTopProcess(ctx, mpool.MustNewZero(), txnClient, sharedTxn, + nil, services[0], nil, nil, nil, nil, nil) + require.NoError(t, lockop.LockTableWithMode(nil, sharedProc, 10, + types.T_int32.ToType(), lockpb.LockMode_Shared, false)) }, nil, ) diff --git a/pkg/sql/compile/operator.go b/pkg/sql/compile/operator.go index 4c6cda80881da..5861b535cfe67 100644 --- a/pkg/sql/compile/operator.go +++ b/pkg/sql/compile/operator.go @@ -787,11 +787,11 @@ 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 { - arg.LockTable(target.TableId, false) + arg.LockTableWithMode(target.TableId, target.Mode, false) } } return arg, nil diff --git a/pkg/sql/compile/remoterun.go b/pkg/sql/compile/remoterun.go index 9e7937f63b7e9..e911f193a5843 100644 --- a/pkg/sql/compile/remoterun.go +++ b/pkg/sql/compile/remoterun.go @@ -942,11 +942,11 @@ 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 { - lockArg.LockTable(target.TableId, target.ChangeDef) + lockArg.LockTableWithMode(target.TableId, target.Mode, target.ChangeDef) } } op = lockArg diff --git a/pkg/sql/compile/remoterun_test.go b/pkg/sql/compile/remoterun_test.go index 52f09f011efca..e0646bef1ec48 100644 --- a/pkg/sql/compile/remoterun_test.go +++ b/pkg/sql/compile/remoterun_test.go @@ -36,6 +36,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/container/vector" "github.com/matrixorigin/matrixone/pkg/defines" mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test" + lockpb "github.com/matrixorigin/matrixone/pkg/pb/lock" "github.com/matrixorigin/matrixone/pkg/pb/pipeline" planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/pb/txn" @@ -359,6 +360,37 @@ func TestRemoteRunOperatorCodecRoundTrip(t *testing.T) { require.IsType(t, &intersectall.IntersectAll{}, restored) require.Equal(t, vm.IntersectAll, restored.OpType()) }) + + t.Run("SharedTableLock", func(t *testing.T) { + original := lockop.NewArgumentByEngine(nil) + original.AddLockTargetWithMode(42, nil, lockpb.LockMode_Shared, 0, + types.T_int64.ToType(), -1, -1, nil, false) + original.LockTableWithMode(42, lockpb.LockMode_Shared, false) + + restored := roundTrip(t, original) + defer restored.Release() + restoredLock, ok := restored.(*lockop.LockOp) + require.True(t, ok) + targets := restoredLock.CopyToPipelineTarget() + require.Len(t, targets, 1) + require.True(t, targets[0].LockTable) + require.Equal(t, lockpb.LockMode_Shared, targets[0].Mode) + }) + + t.Run("SharedRowLock", func(t *testing.T) { + original := lockop.NewArgumentByEngine(nil) + original.AddLockTargetWithMode(43, nil, lockpb.LockMode_Shared, 0, + types.T_int64.ToType(), -1, -1, nil, false) + + restored := roundTrip(t, original) + defer restored.Release() + restoredLock, ok := restored.(*lockop.LockOp) + require.True(t, ok) + targets := restoredLock.CopyToPipelineTarget() + require.Len(t, targets, 1) + require.False(t, targets[0].LockTable) + require.Equal(t, lockpb.LockMode_Shared, targets[0].Mode) + }) } func TestExternalScanParquetRowGroupShardsRoundtrip(t *testing.T) { diff --git a/pkg/sql/plan/bind_insert.go b/pkg/sql/plan/bind_insert.go index 3a9bfbbdcd4b4..7a1b6d52e7e7c 100644 --- a/pkg/sql/plan/bind_insert.go +++ b/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" @@ -1034,17 +1035,62 @@ func (builder *QueryBuilder) appendModernChildFkMarkOks( childColPos func(colName string) int32, ) (int32, []*plan.Expr, error) { selectNode := builder.qry.Nodes[lastNodeID] + inputTypes := make([]plan.Type, len(selectNode.ProjectList)) + for i, expr := range selectNode.ProjectList { + inputTypes[i] = expr.Typ + } id2name := make(map[uint64]string, len(tableDef.Cols)) for _, col := range tableDef.Cols { id2name[col.ColId] = col.Name } - oks := make([]*plan.Expr, 0, len(tableDef.Fkeys)) + nonSelfFks := make([]*plan.ForeignKeyDef, 0, len(tableDef.Fkeys)) for _, fk := range tableDef.Fkeys { - if fk.ForeignTbl == 0 { - continue // self-referencing FK handled post-execution via DetectSql + if fk.ForeignTbl != 0 { + nonSelfFks = append(nonSelfFks, fk) + } + } + if len(nonSelfFks) == 0 { + return lastNodeID, nil, nil + } + + // Evaluate the incoming row image once. Prerequisite steps lock every referenced + // key from this materialized stream before the final step scans any parent table. + sourceSinkID := appendSinkNodeWithTag(builder, bindCtx, lastNodeID, selectTag) + sourceStep := builder.appendStep(sourceSinkID) + appendSourceScan := func() int32 { + return builder.appendTaggedSinkScan(bindCtx, sourceStep, selectTag) + } + + parentColNames := func(parent *plan.TableDef, colIDs []uint64) ([]string, error) { + idToName := make(map[uint64]string, len(parent.Cols)) + for _, col := range parent.Cols { + idToName[col.ColId] = col.Name + } + names := make([]string, len(colIDs)) + for i, id := range colIDs { + var ok bool + if names[i], ok = idToName[id]; !ok { + return nil, moerr.NewInternalErrorf(builder.GetContext(), + "foreign-key parent column %d not found", id) + } + } + return names, nil + } + partsEqual := func(parts, names []string) bool { + if len(parts) != len(names) { + return false } + for i := range parts { + if catalog.ResolveAlias(parts[i]) != names[i] { + return false + } + } + return true + } + + for _, fk := range nonSelfFks { parentObjRef, parentTableDef, err := builder.compCtx.ResolveById(fk.ForeignTbl, bindCtx.snapshot) if err != nil { return 0, nil, err @@ -1052,6 +1098,117 @@ func (builder *QueryBuilder) appendModernChildFkMarkOks( if parentTableDef == nil { return 0, nil, moerr.NewInternalErrorf(builder.GetContext(), "parent table %d not found", fk.ForeignTbl) } + referencedNames, err := parentColNames(parentTableDef, fk.ForeignCols) + if err != nil { + return 0, nil, err + } + childExprs := make([]*plan.Expr, len(fk.Cols)) + for i, childColID := range fk.Cols { + pos := childColPos(id2name[childColID]) + childExprs[i] = &plan.Expr{Typ: inputTypes[pos], Expr: &plan.Expr_Col{Col: &plan.ColRef{ + RelPos: selectTag, ColPos: int32(pos), + }}} + } + + var lockExpr *plan.Expr + var lockTableDef *plan.TableDef + var lockObjRef *plan.ObjectRef + lockTable := false + var pkeyNames []string + if parentTableDef.Pkey != nil { + pkeyNames = parentTableDef.Pkey.Names + if len(pkeyNames) == 0 && parentTableDef.Pkey.PkeyColName != "" { + pkeyNames = []string{parentTableDef.Pkey.PkeyColName} + } + } + if partsEqual(pkeyNames, referencedNames) { + lockTableDef = parentTableDef + lockObjRef = parentObjRef + if len(childExprs) == 1 { + lockExpr = childExprs[0] + } else { + lockExpr, err = BindFuncExprImplByPlanExpr(builder.GetContext(), "serial", childExprs) + if err != nil { + return 0, nil, err + } + } + } else { + var matchedIndex *plan.IndexDef + for _, idxDef := range parentTableDef.Indexes { + if idxDef.Unique && partsEqual(idxDef.Parts, referencedNames) { + matchedIndex = idxDef + break + } + } + if matchedIndex == nil { + // Legacy schemas may reference a non-unique prefix of a composite key. + // Such a reference has no physical point-lock key, so serialize it with + // parent mutations before the validation scan using a shared table lock. + lockTableDef = parentTableDef + lockObjRef = parentObjRef + lockExpr = childExprs[0] + lockTable = true + } else { + lockObjRef, lockTableDef, err = builder.compCtx.ResolveIndexTableByRef( + parentObjRef, matchedIndex.IndexTableName, bindCtx.snapshot) + if err != nil { + return 0, nil, err + } + prefixLengths, err := catalog.IndexPrefixLengthsFromParamsWithError(matchedIndex.IndexAlgoParams) + if err != nil { + return 0, nil, err + } + keyParts := make([]*plan.Expr, len(childExprs)) + for i, expr := range childExprs { + keyParts[i], err = builder.makeIndexPartExprFromInputExpr(expr, referencedNames[i], prefixLengths) + if err != nil { + return 0, nil, err + } + } + if indexTableStoresSerializedKey(matchedIndex) { + lockExpr, err = BindFuncExprImplByPlanExpr(builder.GetContext(), "serial", keyParts) + if err != nil { + return 0, nil, err + } + } else { + lockExpr = keyParts[0] + } + } + } + lockPkPos, lockTyp := getPkPos(lockTableDef, false) + if lockPkPos < 0 { + return 0, nil, moerr.NewInternalErrorf(builder.GetContext(), + "foreign-key lock table %s has no primary key", lockTableDef.Name) + } + lockInputID := appendSourceScan() + lockTag := builder.genNewBindTag() + lockProject := []*plan.Expr{lockExpr} + lockInputID = builder.appendNode(&plan.Node{ + NodeType: plan.Node_PROJECT, Children: []int32{lockInputID}, + ProjectList: lockProject, BindingTags: []int32{lockTag}, + }, bindCtx) + lockKeySinkID := appendSinkNodeWithTag(builder, bindCtx, lockInputID, lockTag) + lockKeyStep := builder.appendStep(lockKeySinkID) + lockInputID = builder.appendTaggedSinkScan(bindCtx, lockKeyStep, lockTag) + lockNodeID := builder.appendNode(&plan.Node{ + NodeType: plan.Node_LOCK_OP, Children: []int32{lockInputID}, TableDef: lockTableDef, + LockTargets: []*plan.LockTarget{{ + TableId: lockTableDef.TblId, ObjRef: lockObjRef, + PrimaryColIdxInBat: 0, PrimaryColRelPos: lockTag, + PrimaryColTyp: lockTyp, Mode: lockpb.LockMode_Shared, LockTable: lockTable, + }}, + }, bindCtx) + builder.appendStep(lockNodeID) + } + + lastNodeID = appendSourceScan() + selectNode = builder.qry.Nodes[lastNodeID] + oks := make([]*plan.Expr, 0, len(nonSelfFks)) + for _, fk := range nonSelfFks { + parentObjRef, parentTableDef, err := builder.compCtx.ResolveById(fk.ForeignTbl, bindCtx.snapshot) + if err != nil { + return 0, nil, err + } parentTag := builder.genNewBindTag() builder.addNameByColRef(parentTag, parentTableDef) diff --git a/pkg/sql/plan/bind_replace.go b/pkg/sql/plan/bind_replace.go index cbdfd82be227c..84cbafac29c5c 100644 --- a/pkg/sql/plan/bind_replace.go +++ b/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" diff --git a/pkg/sql/plan/build.go b/pkg/sql/plan/build.go index 6980478dc913a..efbe4b0043cbb 100644 --- a/pkg/sql/plan/build.go +++ b/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{ diff --git a/pkg/sql/plan/build_test.go b/pkg/sql/plan/build_test.go index 131ad35e34bea..703ffcbac549d 100644 --- a/pkg/sql/plan/build_test.go +++ b/pkg/sql/plan/build_test.go @@ -18,9 +18,11 @@ import ( "bytes" "context" "encoding/json" + "fmt" "os" "strings" "testing" + "time" "github.com/golang/mock/gomock" "github.com/stretchr/testify/assert" @@ -30,6 +32,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/moerr" moruntime "github.com/matrixorigin/matrixone/pkg/common/runtime" "github.com/matrixorigin/matrixone/pkg/container/types" + lockpb "github.com/matrixorigin/matrixone/pkg/pb/lock" "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect/mysql" "github.com/matrixorigin/matrixone/pkg/sql/parsers/tree" @@ -3049,6 +3052,776 @@ func TestReplaceDetectSqlsMultipleRows(t *testing.T) { assert.Contains(t, preCheck, "3", "pre-check IN list should contain row 3's PK") } +func TestReplaceParentSideFKRestrict(t *testing.T) { + mock := NewMockOptimizer(true) + + // REPLACE on a parent table whose PK is referenced by a child with + // ON DELETE RESTRICT must generate a REPLACE_PARENT_CHK: pre-check SQL + // against the child table (issue #24951, 3.2 RESTRICT case). + logicPlan, err := runOneStmt(mock, t, "REPLACE INTO replace_fk_p VALUES (1, 'p1_new')") + if err != nil { + t.Fatalf("%+v", err) + } + + query := logicPlan.GetQuery() + assert.NotNil(t, query) + require.NotEmpty(t, query.DetectSqls) + assert.True(t, strings.HasPrefix(query.DetectSqls[0], "REPLACE_PARENT_LOCK:"), + "parent row lock must run before the RESTRICT check") + + var preCheck string + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") { + preCheck = strings.TrimPrefix(sql, "REPLACE_PARENT_CHK:") + break + } + } + assert.NotEmpty(t, preCheck, + "RESTRICT parent-side FK REPLACE should generate a REPLACE_PARENT_CHK: pre-check SQL") + assert.Contains(t, preCheck, "replace_fk_c", "pre-check SQL should target the child table") + assert.Contains(t, preCheck, "`pid`", "pre-check SQL should reference the child FK column") + assert.Contains(t, preCheck, "`id` = cast(1 as INT)", "pre-check SQL should assignment-cast the supplied PK value") + assert.Contains(t, preCheck, "exists (select 1", "pre-check should resolve the actual conflicting parent row") + + // No CASCADE/SET NULL action SQL should be produced for RESTRICT. + for _, sql := range query.DetectSqls { + assert.False(t, strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:"), + "RESTRICT parent-side FK must NOT generate an action SQL, got: %s", sql) + } +} + +func TestReplaceParentSideFKCascade(t *testing.T) { + mock := NewMockOptimizer(true) + + // REPLACE on a parent table whose PK is referenced by a child with + // ON DELETE CASCADE must generate a REPLACE_PARENT_ACTION: delete SQL + // against the child table (issue #24951, 3.2 CASCADE case). + logicPlan, err := runOneStmt(mock, t, "REPLACE INTO replace_fk_cp VALUES (1, 'p1_new')") + if err != nil { + t.Fatalf("%+v", err) + } + + query := logicPlan.GetQuery() + assert.NotNil(t, query) + require.NotEmpty(t, query.DetectSqls) + assert.True(t, strings.HasPrefix(query.DetectSqls[0], "REPLACE_PARENT_LOCK:"), + "parent row lock must run before the CASCADE action") + + var action string + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:") { + action = strings.TrimPrefix(sql, "REPLACE_PARENT_ACTION:") + break + } + } + assert.NotEmpty(t, action, + "CASCADE parent-side FK REPLACE should generate a REPLACE_PARENT_ACTION: SQL") + assert.Contains(t, action, "delete from", "CASCADE action should be a DELETE on the child") + assert.Contains(t, action, "replace_fk_cc", "CASCADE action should target the child table") + assert.Contains(t, action, "`pid`", "CASCADE action should filter on the child FK column") + assert.Contains(t, action, "`id` = cast(1 as INT)", "CASCADE action should assignment-cast the supplied PK value") + assert.Contains(t, action, "exists (select 1", "CASCADE should resolve the actual conflicting parent row") + + // CASCADE must NOT generate a parent-child RESTRICT pre-check. + for _, sql := range query.DetectSqls { + assert.False(t, strings.HasPrefix(sql, "REPLACE_PARENT_CHK:"), + "CASCADE parent-side FK must NOT generate a pre-check SQL, got: %s", sql) + } +} + +func TestReplaceParentSideFKExplicitColumns(t *testing.T) { + mock := NewMockOptimizer(true) + + // Explicit column list (mixed case) must still resolve the PK position and + // generate the parent-side pre-check. + logicPlan, err := runOneStmt(mock, t, + "REPLACE INTO replace_fk_p (ID, V) VALUES (1, 'p1_new')") + if err != nil { + t.Fatalf("%+v", err) + } + + query := logicPlan.GetQuery() + assert.NotNil(t, query) + + var preCheck string + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") { + preCheck = strings.TrimPrefix(sql, "REPLACE_PARENT_CHK:") + break + } + } + assert.NotEmpty(t, preCheck, + "parent-side pre-check should be generated with an explicit, mixed-case column list") + assert.Contains(t, preCheck, "replace_fk_c", "pre-check SQL should target the child table") + assert.Contains(t, preCheck, "`pid`", "pre-check SQL should reference the child FK column") + assert.Contains(t, preCheck, "`id` = cast(1 as INT)", "pre-check SQL should assignment-cast the supplied PK value") +} + +func TestReplaceParentSideFKNoAction(t *testing.T) { + mock := NewMockOptimizer(true) + + // ON DELETE NO ACTION behaves like RESTRICT: it must generate a + // REPLACE_PARENT_CHK: pre-check, not a CASCADE/SET NULL action. + logicPlan, err := runOneStmt(mock, t, "REPLACE INTO replace_fk_np VALUES (1, 'p1_new')") + if err != nil { + t.Fatalf("%+v", err) + } + + query := logicPlan.GetQuery() + assert.NotNil(t, query) + + var preCheck string + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") { + preCheck = strings.TrimPrefix(sql, "REPLACE_PARENT_CHK:") + break + } + } + assert.NotEmpty(t, preCheck, + "NO ACTION parent-side FK REPLACE should generate a REPLACE_PARENT_CHK: pre-check SQL") + assert.Contains(t, preCheck, "replace_fk_nc", "pre-check SQL should target the child table") + for _, sql := range query.DetectSqls { + assert.False(t, strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:"), + "NO ACTION parent-side FK must NOT generate an action SQL, got: %s", sql) + } +} + +func TestReplaceParentSideFKSetDefault(t *testing.T) { + mock := NewMockOptimizer(true) + + logicPlan, err := runOneStmt(mock, t, "REPLACE INTO replace_fk_dp VALUES (1, 'p1_new')") + if err != nil { + t.Fatalf("%+v", err) + } + + query := logicPlan.GetQuery() + assert.NotNil(t, query) + + var preCheck string + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_CHK:") { + preCheck = strings.TrimPrefix(sql, "REPLACE_PARENT_CHK:") + break + } + } + assert.NotEmpty(t, preCheck, + "SET DEFAULT parent-side FK REPLACE should generate a REPLACE_PARENT_CHK: pre-check SQL") + assert.Contains(t, preCheck, "replace_fk_dc", "pre-check SQL should target the child table") + for _, sql := range query.DetectSqls { + assert.False(t, strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:"), + "SET DEFAULT parent-side FK must NOT generate an action SQL, got: %s", sql) + } +} + +func TestReplaceParentSideFKMultiRow(t *testing.T) { + mock := NewMockOptimizer(true) + + // Multi-row REPLACE: every literal PK value must be embedded into the same + // parent-side action IN list (issue #24951 data-integrity case). + logicPlan, err := runOneStmt(mock, t, + "REPLACE INTO replace_fk_cp VALUES (1, 'a'), (2, 'b')") + if err != nil { + t.Fatalf("%+v", err) + } + + query := logicPlan.GetQuery() + assert.NotNil(t, query) + + var action string + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:") { + action = strings.TrimPrefix(sql, "REPLACE_PARENT_ACTION:") + break + } + } + assert.NotEmpty(t, action, + "multi-row CASCADE parent-side REPLACE should generate an action SQL") + assert.Contains(t, action, "1", "action IN list should contain row 1's PK") + assert.Contains(t, action, "2", "action IN list should contain row 2's PK") +} + +func TestReplaceParentSideFKMixedLiteralRows(t *testing.T) { + mock := NewMockOptimizer(true) + + // A mixed VALUES list cannot partially apply parent-side actions because that + // would leave the non-literal rows unchecked. + logicPlan, err := runOneStmt(mock, t, + "REPLACE INTO replace_fk_cp VALUES (1, 'a'), (rand(), 'b')") + assert.Error(t, err) + assert.Nil(t, logicPlan) +} + +func TestReplaceParentSideFKSetNull(t *testing.T) { + mock := NewMockOptimizer(true) + + // REPLACE on a parent table whose PK is referenced by a child with + // ON DELETE SET NULL must generate a REPLACE_PARENT_ACTION: update SQL + // that nulls the child FK column. + logicPlan, err := runOneStmt(mock, t, "REPLACE INTO replace_fk_sp VALUES (1, 'p1_new')") + if err != nil { + t.Fatalf("%+v", err) + } + + query := logicPlan.GetQuery() + assert.NotNil(t, query) + + var action string + for _, sql := range query.DetectSqls { + if strings.HasPrefix(sql, "REPLACE_PARENT_ACTION:") { + action = strings.TrimPrefix(sql, "REPLACE_PARENT_ACTION:") + break + } + } + assert.NotEmpty(t, action, + "SET NULL parent-side FK REPLACE should generate a REPLACE_PARENT_ACTION: SQL") + assert.Contains(t, action, "update", "SET NULL action should be an UPDATE on the child") + assert.Contains(t, action, "replace_fk_sc", "SET NULL action should target the child table") + assert.Contains(t, action, "null", "SET NULL action should set the child FK column to null") + assert.Contains(t, action, "`id` = cast(1 as INT)", "SET NULL action should assignment-cast the supplied PK value") + + // SET NULL must NOT generate a parent-child RESTRICT pre-check. + for _, sql := range query.DetectSqls { + assert.False(t, strings.HasPrefix(sql, "REPLACE_PARENT_CHK:"), + "SET NULL parent-side FK must NOT generate a pre-check SQL, got: %s", sql) + } +} + +func TestReplaceParentSideFKNonLiteralSkip(t *testing.T) { + mock := NewMockOptimizer(true) + + // Non-literal PK expressions cannot be safely embedded into the background + // parent-side SQL, so the statement must fail closed. + logicPlan, err := runOneStmt(mock, t, "REPLACE INTO replace_fk_p VALUES (rand(), 'x')") + assert.Error(t, err) + assert.Nil(t, logicPlan) +} + +func TestReplaceParentSideFKUnsupportedSources(t *testing.T) { + mock := NewMockOptimizer(true) + for _, sql := range []string{ + "REPLACE INTO replace_fk_p VALUES (?, 'x')", + "REPLACE INTO replace_fk_p SELECT deptno, dname FROM dept", + } { + logicPlan, err := runOneStmt(mock, t, sql) + assert.Error(t, err, sql) + assert.Nil(t, logicPlan, sql) + } +} + +func TestReplaceParentSideFKUniquePrefixConflict(t *testing.T) { + mock := NewMockOptimizer(true) + parent := DeepCopyTableDef(mock.ctxt.tables["replace_fk_cp"], true) + parent.Indexes = append(parent.Indexes, &plan.IndexDef{ + Unique: true, + Parts: []string{"v"}, + IndexAlgoParams: `{"prefix_lengths":"v:4"}`, + }) + stmt, err := mysql.ParseOne(context.Background(), + "REPLACE INTO replace_fk_cp VALUES (2, 'abcdyyyy')", 1) + require.NoError(t, err) + + _, _, actions, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, stmt.(*tree.Replace)) + require.NoError(t, err) + require.Len(t, actions, 1) + assert.Contains(t, actions[0], "substring(`__mo_replace_parent`.`v`, 1, 4)") + assert.Contains(t, actions[0], `substring(cast("abcdyyyy" as VARCHAR(20)), 1, 4)`) +} + +func TestReplaceParentSideFKAssignmentCastAndLock(t *testing.T) { + mock := NewMockOptimizer(true) + parent := DeepCopyTableDef(mock.ctxt.tables["replace_fk_cp"], true) + for _, col := range parent.Cols { + if col.Name == "v" { + col.Typ = plan.Type{Id: int32(types.T_decimal64), Width: 5, Scale: 2} + } + } + parent.Indexes = append(parent.Indexes, &plan.IndexDef{Unique: true, Parts: []string{"v"}}) + stmt, err := mysql.ParseOne(context.Background(), + "REPLACE INTO replace_fk_cp VALUES (2, 1.234)", 1) + require.NoError(t, err) + + lockSQL, _, actions, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, stmt.(*tree.Replace)) + require.NoError(t, err) + require.Len(t, actions, 1) + assert.Contains(t, lockSQL, "`__mo_replace_parent`.`v` = cast(1.234 as DECIMAL(5,2))") + assert.Contains(t, lockSQL, "for update") + assert.Contains(t, actions[0], "`__mo_replace_parent`.`v` = cast(1.234 as DECIMAL(5,2))") +} + +func TestReplaceParentSideFKRejectsOverWidthConflictLiteral(t *testing.T) { + mock := NewMockOptimizer(true) + parent := DeepCopyTableDef(mock.ctxt.tables["replace_fk_cp"], true) + for _, col := range parent.Cols { + if col.Name == "v" { + col.Typ = plan.Type{Id: int32(types.T_varchar), Width: 3} + } + } + parent.Indexes = append(parent.Indexes, &plan.IndexDef{Unique: true, Parts: []string{"v"}}) + stmt, err := mysql.ParseOne(context.Background(), + "REPLACE INTO replace_fk_cp VALUES (2, 'abcd')", 1) + require.NoError(t, err) + + lockSQL, checks, actions, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, stmt.(*tree.Replace)) + require.ErrorContains(t, err, "larger than Dest length") + assert.Empty(t, lockSQL) + assert.Empty(t, checks) + assert.Empty(t, actions) +} + +func TestChildInsertLocksForeignKeyParentShared(t *testing.T) { + mock := NewMockOptimizer(true) + logicPlan, err := runOneStmt(mock, t, "INSERT INTO replace_fk_c VALUES (10, 1)") + require.NoError(t, err) + + parentID := mock.ctxt.tables["replace_fk_p"].TblId + query := logicPlan.GetQuery() + found := false + lockNodeID := int32(-1) + parentScanIDs := make([]int32, 0, 1) + for nodeID, node := range query.Nodes { + if node.NodeType == plan.Node_TABLE_SCAN && node.TableDef != nil && node.TableDef.TblId == parentID { + assert.Empty(t, node.LockTargets, "the raw parent scan must not carry a shared lock") + parentScanIDs = append(parentScanIDs, int32(nodeID)) + } + for _, target := range node.LockTargets { + if target.TableId == parentID && target.Mode == lockpb.LockMode_Shared { + found = true + lockNodeID = int32(nodeID) + assert.Equal(t, int32(0), target.PrimaryColRelPos) + require.Len(t, node.Children, 1) + lockInput := query.Nodes[node.Children[0]] + require.Less(t, int(target.PrimaryColIdxInBat), len(lockInput.ProjectList)) + assert.Equal(t, target.PrimaryColTyp.Id, lockInput.ProjectList[target.PrimaryColIdxInBat].Typ.Id) + child := lockInput + for child.NodeType != plan.Node_SINK_SCAN && len(child.Children) == 1 { + child = query.Nodes[child.Children[0]] + } + assert.Equal(t, plan.Node_SINK_SCAN, child.NodeType, + "the prerequisite lock must read the materialized child row image") + } + } + } + assert.True(t, found, "child FK validation must hold a shared lock on its parent row") + require.NotEmpty(t, parentScanIDs) + stepContaining := func(target int32) int { + var contains func(int32) bool + contains = func(nodeID int32) bool { + if nodeID == target { + return true + } + for _, childID := range query.Nodes[nodeID].Children { + if contains(childID) { + return true + } + } + return false + } + for step, rootID := range query.Steps { + if contains(rootID) { + return step + } + } + return -1 + } + lockStep := stepContaining(lockNodeID) + require.GreaterOrEqual(t, lockStep, 0) + for _, scanID := range parentScanIDs { + assert.Greater(t, stepContaining(scanID), lockStep, + "the referenced-key lock step must finish before the parent scan step starts") + } +} + +func TestChildInsertLocksCompositeParentPrimaryKey(t *testing.T) { + mock := NewMockOptimizer(true) + parent := mock.ctxt.tables["replace_fk_p"] + child := mock.ctxt.tables["replace_fk_c"] + parent.Cols = append(parent.Cols, + &plan.ColDef{Name: "k", ColId: 3, Typ: plan.Type{Id: int32(types.T_int32), Width: 32}}, + &plan.ColDef{Name: catalog.CPrimaryKeyColName, ColId: 4, Hidden: true, + Typ: plan.Type{Id: int32(types.T_varchar), Width: 65535}}, + ) + parent.Pkey = &plan.PrimaryKeyDef{Names: []string{"id", "k"}, PkeyColName: catalog.CPrimaryKeyColName} + if parent.Name2ColIndex == nil { + parent.Name2ColIndex = make(map[string]int32, len(parent.Cols)) + for i, col := range parent.Cols { + parent.Name2ColIndex[col.Name] = int32(i) + } + } + parent.Name2ColIndex["k"] = int32(len(parent.Cols) - 2) + parent.Name2ColIndex[catalog.CPrimaryKeyColName] = int32(len(parent.Cols) - 1) + child.Fkeys[0].Cols = []uint64{0, 1} + child.Fkeys[0].ForeignCols = []uint64{0, 3} + + logicPlan, err := runOneStmt(mock, t, "INSERT INTO replace_fk_c VALUES (10, 1)") + require.NoError(t, err) + for _, node := range logicPlan.GetQuery().Nodes { + for _, target := range node.LockTargets { + if target.TableId != parent.TblId || target.Mode != lockpb.LockMode_Shared { + continue + } + lockInput := logicPlan.GetQuery().Nodes[node.Children[0]] + require.Less(t, int(target.PrimaryColIdxInBat), len(lockInput.ProjectList), + "lock input=%+v target=%+v", lockInput, target) + assert.Equal(t, target.PrimaryColTyp.Id, lockInput.ProjectList[target.PrimaryColIdxInBat].Typ.Id) + assert.Equal(t, int32(types.T_varchar), target.PrimaryColTyp.Id) + return + } + } + t.Fatal("composite parent primary key shared lock not found") +} + +func TestChildInsertLocksCompositeParentPrimaryKeyPrefixTable(t *testing.T) { + mock := NewMockOptimizer(true) + parent := mock.ctxt.tables["replace_fk_p"] + parent.Cols = append(parent.Cols, + &plan.ColDef{Name: "k", ColId: 3, Typ: plan.Type{Id: int32(types.T_int32), Width: 32}}, + &plan.ColDef{Name: catalog.CPrimaryKeyColName, ColId: 4, Hidden: true, + Typ: plan.Type{Id: int32(types.T_varchar), Width: 65535}}, + ) + parent.Pkey = &plan.PrimaryKeyDef{Names: []string{"id", "k"}, PkeyColName: catalog.CPrimaryKeyColName} + if parent.Name2ColIndex == nil { + parent.Name2ColIndex = make(map[string]int32, len(parent.Cols)) + for i, col := range parent.Cols { + parent.Name2ColIndex[col.Name] = int32(i) + } + } + parent.Name2ColIndex["k"] = int32(len(parent.Cols) - 2) + parent.Name2ColIndex[catalog.CPrimaryKeyColName] = int32(len(parent.Cols) - 1) + + logicPlan, err := runOneStmt(mock, t, "INSERT INTO replace_fk_c VALUES (10, 1)") + require.NoError(t, err) + query := logicPlan.GetQuery() + stepContaining := func(target int32) int { + var contains func(int32) bool + contains = func(nodeID int32) bool { + if nodeID == target { + return true + } + for _, childID := range query.Nodes[nodeID].Children { + if contains(childID) { + return true + } + } + return false + } + for step, rootID := range query.Steps { + if contains(rootID) { + return step + } + } + return -1 + } + foundParentScan := false + for _, node := range query.Nodes { + if node.NodeType == plan.Node_TABLE_SCAN && node.TableDef != nil && node.TableDef.TblId == parent.TblId { + foundParentScan = true + } + } + for nodeID, node := range query.Nodes { + for _, target := range node.LockTargets { + if target.TableId != parent.TblId || target.Mode != lockpb.LockMode_Shared { + continue + } + assert.True(t, target.LockTable) + lockStep := stepContaining(int32(nodeID)) + require.GreaterOrEqual(t, lockStep, 0) + assert.Less(t, lockStep, len(query.Steps)-1) + assert.True(t, foundParentScan) + return + } + } + t.Fatal("composite parent primary-key prefix shared table lock not found") +} + +func TestChildInsertLocksReferencedUniqueIndexKey(t *testing.T) { + mock := NewMockOptimizer(true) + parent := mock.ctxt.tables["replace_fk_p"] + child := mock.ctxt.tables["replace_fk_c"] + child.Cols[1].Typ = plan.Type{Id: int32(types.T_varchar), Width: 20} + child.Fkeys[0].ForeignCols = []uint64{1} + indexName := "__mo_index_fk_parent_v" + indexID := uint64(77901) + parent.Indexes = append(parent.Indexes, &plan.IndexDef{ + IndexName: "uk_v", IndexTableName: indexName, Parts: []string{"v"}, + Unique: true, TableExist: true, IndexAlgo: catalog.MoIndexDefaultAlgo.ToString(), + }) + indexTable := &plan.TableDef{ + TblId: indexID, Name: indexName, + Cols: []*plan.ColDef{ + {Name: catalog.IndexTableIndexColName, ColId: 0, Typ: plan.Type{Id: int32(types.T_varchar), Width: 20}}, + {Name: catalog.Row_ID, ColId: 1, Hidden: true, Typ: plan.Type{Id: int32(types.T_Rowid)}}, + }, + Pkey: &plan.PrimaryKeyDef{Names: []string{catalog.IndexTableIndexColName}, + PkeyColName: catalog.IndexTableIndexColName}, + Name2ColIndex: map[string]int32{catalog.IndexTableIndexColName: 0, catalog.Row_ID: 1}, + } + mock.ctxt.tables[indexName] = indexTable + mock.ctxt.objects[indexName] = &plan.ObjectRef{ + Obj: int64(indexID), SchemaName: mock.ctxt.objects["replace_fk_p"].SchemaName, ObjName: indexName, + } + + logicPlan, err := runOneStmt(mock, t, "INSERT INTO replace_fk_c VALUES (10, 'x')") + require.NoError(t, err) + for _, node := range logicPlan.GetQuery().Nodes { + for _, target := range node.LockTargets { + if target.TableId == indexID && target.Mode == lockpb.LockMode_Shared { + assert.Equal(t, int32(types.T_varchar), target.PrimaryColTyp.Id) + return + } + } + } + t.Fatal("referenced unique-index shared lock not found") +} + +func TestDeepCopyPreservesSharedLockMode(t *testing.T) { + target := &plan.LockTarget{ + TableId: 42, + ObjRef: &plan.ObjectRef{Obj: 42, ObjName: "parent"}, + Mode: lockpb.LockMode_Shared, + PrimaryColRelPos: 11, + FilterColRelPos: 12, + PartitionColIdxInBat: 13, + HasPartitionCol: true, + LockRows: makePlan2Int64ConstExprWithType(7), + } + assertScalarFields := func(t *testing.T, copied *plan.LockTarget) { + t.Helper() + assert.Equal(t, lockpb.LockMode_Shared, copied.Mode) + assert.Equal(t, int32(11), copied.PrimaryColRelPos) + assert.Equal(t, int32(12), copied.FilterColRelPos) + assert.Equal(t, int32(13), copied.PartitionColIdxInBat) + assert.True(t, copied.HasPartitionCol) + } + + direct := DeepCopyLockTarget(target) + require.NotSame(t, target, direct) + assertScalarFields(t, direct) + require.NotSame(t, target.ObjRef, direct.ObjRef) + require.NotSame(t, target.LockRows, direct.LockRows) + + node := &plan.Node{NodeType: plan.Node_LOCK_OP, LockTargets: []*plan.LockTarget{target}} + nodeCopy := DeepCopyNode(node) + require.Len(t, nodeCopy.LockTargets, 1) + assertScalarFields(t, nodeCopy.LockTargets[0]) + require.NotSame(t, target, nodeCopy.LockTargets[0]) + + queryCopy := DeepCopyQuery(&plan.Query{Nodes: []*plan.Node{node}}) + require.Len(t, queryCopy.Nodes, 1) + require.Len(t, queryCopy.Nodes[0].LockTargets, 1) + assertScalarFields(t, queryCopy.Nodes[0].LockTargets[0]) + require.NotSame(t, target, queryCopy.Nodes[0].LockTargets[0]) +} + +func TestReplaceParentSideFKOmittedUniqueDefaults(t *testing.T) { + parseReplace := func(t *testing.T, sql string) *tree.Replace { + t.Helper() + stmt, err := mysql.ParseOne(context.Background(), sql, 1) + require.NoError(t, err) + return stmt.(*tree.Replace) + } + newParent := func(t *testing.T) (*MockOptimizer, *plan.TableDef) { + t.Helper() + mock := NewMockOptimizer(true) + parent := DeepCopyTableDef(mock.ctxt.tables["replace_fk_cp"], true) + parent.Indexes = append(parent.Indexes, &plan.IndexDef{ + Unique: true, + Parts: []string{"v"}, + }) + return mock, parent + } + parentCol := func(t *testing.T, parent *plan.TableDef, name string) *plan.ColDef { + t.Helper() + for _, col := range parent.Cols { + if col.Name == name { + return col + } + } + t.Fatalf("missing parent column %s", name) + return nil + } + + t.Run("nullable default cannot conflict", func(t *testing.T) { + mock, parent := newParent(t) + v := parentCol(t, parent, "v") + v.Default = &plan.Default{NullAbility: true} + + _, _, actions, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, + parseReplace(t, "REPLACE INTO replace_fk_cp(id) VALUES (1)")) + require.NoError(t, err) + require.Len(t, actions, 1) + assert.Contains(t, actions[0], "`__mo_replace_parent`.`id` = cast(1 as INT)") + assert.NotContains(t, actions[0], "`__mo_replace_parent`.`v` =") + }) + + t.Run("constant prefix default participates", func(t *testing.T) { + mock, parent := newParent(t) + parent.Indexes[len(parent.Indexes)-1].IndexAlgoParams = `{"prefix_lengths":"v:4"}` + v := parentCol(t, parent, "v") + v.Default = &plan.Default{ + NullAbility: true, + Expr: makeStringConstExpr(v.Typ, "abcdyyyy"), + } + + _, _, actions, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, + parseReplace(t, "REPLACE INTO replace_fk_cp(id) VALUES (1)")) + require.NoError(t, err) + require.Len(t, actions, 1) + assert.Contains(t, actions[0], "substring(`__mo_replace_parent`.`v`, 1, 4)") + assert.Contains(t, actions[0], `substring(cast("abcdyyyy" as VARCHAR(20)), 1, 4)`) + }) + + t.Run("dynamic default fails closed", func(t *testing.T) { + mock, parent := newParent(t) + v := parentCol(t, parent, "v") + v.Default = &plan.Default{ + NullAbility: true, + Expr: &plan.Expr{ + Typ: v.Typ, + Expr: &plan.Expr_Col{Col: &plan.ColRef{Name: "dynamic_default"}}, + }, + } + + _, _, _, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, + parseReplace(t, "REPLACE INTO replace_fk_cp(id) VALUES (1)")) + require.ErrorContains(t, err, "non-literal default conflict key") + }) + + t.Run("numeric literal defaults participate", func(t *testing.T) { + cases := []struct { + name string + typ plan.Type + literal *plan.Expr_Lit + expected string + }{ + {name: "int8", typ: plan.Type{Id: int32(types.T_int8)}, literal: makePlan2Int8ConstExpr(-8), expected: "cast(-8 as TINYINT)"}, + {name: "int16", typ: plan.Type{Id: int32(types.T_int16)}, literal: makePlan2Int16ConstExpr(-16), expected: "cast(-16 as SMALLINT)"}, + {name: "int32", typ: plan.Type{Id: int32(types.T_int32)}, literal: makePlan2Int32ConstExpr(-32), expected: "cast(-32 as INT)"}, + {name: "int64", typ: plan.Type{Id: int32(types.T_int64)}, literal: makePlan2Int64ConstExpr(-64), expected: "cast(-64 as BIGINT)"}, + {name: "uint8", typ: plan.Type{Id: int32(types.T_uint8)}, literal: makePlan2Uint8ConstExpr(8), expected: "cast(8 as TINYINT UNSIGNED)"}, + {name: "uint16", typ: plan.Type{Id: int32(types.T_uint16)}, literal: makePlan2Uint16ConstExpr(16), expected: "cast(16 as SMALLINT UNSIGNED)"}, + {name: "uint32", typ: plan.Type{Id: int32(types.T_uint32)}, literal: makePlan2Uint32ConstExpr(32), expected: "cast(32 as INT UNSIGNED)"}, + {name: "uint64", typ: plan.Type{Id: int32(types.T_uint64)}, literal: makePlan2Uint64ConstExpr(64), expected: "cast(64 as BIGINT UNSIGNED)"}, + {name: "float32", typ: plan.Type{Id: int32(types.T_float32)}, literal: makePlan2Float32ConstExpr(1.25), expected: "cast(1.25 as FLOAT)"}, + {name: "float64", typ: plan.Type{Id: int32(types.T_float64)}, literal: makePlan2Float64ConstExpr(2.5), expected: "cast(2.5 as DOUBLE)"}, + {name: "bool", typ: plan.Type{Id: int32(types.T_bool)}, literal: makePlan2BoolConstExpr(true), expected: "cast(true as BOOL)"}, + { + name: "enum", typ: plan.Type{Id: int32(types.T_enum), Enumvalues: "small,medium,large"}, + literal: &plan.Expr_Lit{Lit: &plan.Literal{Value: &plan.Literal_EnumVal{EnumVal: 2}}}, + expected: `cast(2 as ENUM("small","medium","large"))`, + }, + { + name: "decimal64", typ: plan.Type{Id: int32(types.T_decimal64), Width: 5, Scale: 2}, + literal: &plan.Expr_Lit{Lit: &plan.Literal{Value: &plan.Literal_Decimal64Val{ + Decimal64Val: &plan.Decimal64{A: 123}, + }}}, expected: "cast(1.23 as DECIMAL(5,2))", + }, + { + name: "decimal128", typ: plan.Type{Id: int32(types.T_decimal128), Width: 20, Scale: 2}, + literal: &plan.Expr_Lit{Lit: &plan.Literal{Value: &plan.Literal_Decimal128Val{ + Decimal128Val: &plan.Decimal128{A: 123}, + }}}, expected: "cast(1.23 as DECIMAL(20,2))", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + mock, parent := newParent(t) + v := parentCol(t, parent, "v") + v.Typ = tc.typ + v.Default = &plan.Default{NullAbility: false, Expr: &plan.Expr{Typ: tc.typ, Expr: tc.literal}} + + lockSQL, _, actions, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, + parseReplace(t, "REPLACE INTO replace_fk_cp(id) VALUES (1)")) + require.NoError(t, err) + require.Len(t, actions, 1) + assert.Contains(t, lockSQL, tc.expected) + assert.Contains(t, actions[0], tc.expected) + _, err = mysql.ParseOne(context.Background(), lockSQL, 1) + require.NoError(t, err, "generated parent lock SQL must be parseable") + }) + } + }) + + t.Run("temporal defaults participate", func(t *testing.T) { + dateValue, err := types.ParseDateCast("2026-07-15") + require.NoError(t, err) + timeValue, err := types.ParseTime("12:34:56.123", 3) + require.NoError(t, err) + datetimeValue, err := types.ParseDatetime("2026-07-15 12:34:56.123", 3) + require.NoError(t, err) + + location := time.UTC + mockForLocation := NewMockOptimizer(true) + if sessionLocation := mockForLocation.ctxt.GetProcess().GetSessionInfo().TimeZone; sessionLocation != nil { + location = sessionLocation + } + timestampValue, err := types.ParseTimestamp(location, "2026-07-15 12:34:56.123", 3) + require.NoError(t, err) + + cases := []struct { + name string + typ plan.Type + literal *plan.Expr_Lit + expected string + }{ + { + name: "date", + typ: plan.Type{Id: int32(types.T_date)}, + literal: makePlan2DateConstExpr(int32(dateValue)), + expected: `"2026-07-15"`, + }, + { + name: "time", + typ: plan.Type{Id: int32(types.T_time), Scale: 3}, + literal: makePlan2TimeConstExpr(int64(timeValue)), + expected: `"12:34:56.123"`, + }, + { + name: "datetime", + typ: plan.Type{Id: int32(types.T_datetime), Scale: 3}, + literal: makePlan2DateTimeConstExpr(int64(datetimeValue)), + expected: `"2026-07-15 12:34:56.123"`, + }, + { + name: "timestamp", + typ: plan.Type{Id: int32(types.T_timestamp), Scale: 3}, + literal: makePlan2TimestampConstExpr(int64(timestampValue)), + expected: `"2026-07-15 12:34:56.123"`, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + mock, parent := newParent(t) + v := parentCol(t, parent, "v") + v.Typ = tc.typ + v.Default = &plan.Default{ + NullAbility: true, + Expr: &plan.Expr{Typ: tc.typ, Expr: tc.literal}, + } + + _, _, actions, err := genParentSideReplaceFKSqls( + &mock.ctxt, mock.ctxt.objects["replace_fk_cp"], parent, + parseReplace(t, "REPLACE INTO replace_fk_cp(id) VALUES (1)")) + require.NoError(t, err) + require.Len(t, actions, 1) + expectedType := strings.ToUpper(types.T(tc.typ.Id).String()) + if tc.typ.Scale > 0 && types.T(tc.typ.Id) != types.T_date { + expectedType += fmt.Sprintf("(%d)", tc.typ.Scale) + } + assert.Contains(t, actions[0], "`__mo_replace_parent`.`v` = cast("+tc.expected+" as "+expectedType+")") + }) + } + }) +} + func TestReplaceODKU(t *testing.T) { mock := NewMockOptimizer(true) // INSERT ON DUPLICATE KEY UPDATE should be rewritten to REPLACE path diff --git a/pkg/sql/plan/build_util.go b/pkg/sql/plan/build_util.go index 60e17167805a5..9d98d8d1801a2 100644 --- a/pkg/sql/plan/build_util.go +++ b/pkg/sql/plan/build_util.go @@ -18,7 +18,10 @@ import ( "context" "fmt" "regexp" + "strconv" "strings" + "time" + "unicode/utf8" "github.com/matrixorigin/matrixone/pkg/catalog" "github.com/matrixorigin/matrixone/pkg/common/moerr" @@ -1188,6 +1191,345 @@ func genPreCheckSqlsForReplaceFKSelfRefer( return ret, nil } +func genParentSideReplaceFKSqls( + ctx CompilerContext, + parentRef *plan.ObjectRef, + parent *plan.TableDef, + stmt *tree.Replace, +) (string, []string, []string, error) { + if stmt.Rows == nil || len(parent.RefChildTbls) == 0 { + return "", nil, nil, nil + } + hasNonSelfReference := false + for _, childID := range parent.RefChildTbls { + if childID != 0 { + hasNonSelfReference = true + break + } + } + if !hasNonSelfReference { + return "", nil, nil, nil + } + values, ok := stmt.Rows.Select.(*tree.ValuesClause) + if !ok { + return "", nil, nil, moerr.NewNotSupported(ctx.GetContext(), "REPLACE SELECT/TABLE on a referenced parent table") + } + positions := make(map[string]int) + if len(stmt.Columns) > 0 { + for i, col := range stmt.Columns { + positions[strings.ToLower(string(col))] = i + } + } else { + pos := 0 + for _, col := range parent.Cols { + if !col.Hidden { + positions[col.Name] = pos + pos++ + } + } + } + quoteIdentifier := func(name string) string { + return "`" + strings.ReplaceAll(name, "`", "``") + "`" + } + qualifiedCol := func(alias, name string) string { + return quoteIdentifier(alias) + "." + quoteIdentifier(name) + } + findParentCol := func(name string) (*plan.ColDef, bool) { + if pos, found := parent.Name2ColIndex[name]; found && int(pos) < len(parent.Cols) { + return parent.Cols[pos], true + } + for _, col := range parent.Cols { + if col.Name == name { + return col, true + } + } + return nil, false + } + + type uniqueKey struct { + parts []string + prefixLengths map[string]int + } + uniqueKeys := make([]uniqueKey, 0, 1+len(parent.Indexes)) + if parent.Pkey != nil && len(parent.Pkey.Names) > 0 { + uniqueKeys = append(uniqueKeys, uniqueKey{parts: parent.Pkey.Names}) + } + for _, idx := range parent.Indexes { + if !idx.Unique { + continue + } + parts := make([]string, len(idx.Parts)) + for i, part := range idx.Parts { + parts[i] = catalog.ResolveAlias(part) + } + prefixLengths, err := catalog.IndexPrefixLengthsFromParamsWithError(idx.IndexAlgoParams) + if err != nil { + return "", nil, nil, err + } + uniqueKeys = append(uniqueKeys, uniqueKey{parts: parts, prefixLengths: prefixLengths}) + } + + const parentAlias = "__mo_replace_parent" + literalFmt := tree.NewFmtCtx(dialect.MYSQL, tree.WithQuoteString(true)) + formatStringLiteral := func(value string) string { + tree.NewNumVal(value, value, false, tree.P_char).Format(literalFmt) + formatted := literalFmt.String() + literalFmt.Reset() + return formatted + } + formatLiteral := func(expr *plan.Expr) (string, bool, bool) { + lit := expr.GetLit() + if lit == nil { + return "", false, false + } + if lit.Isnull { + return "", true, true + } + switch val := lit.Value.(type) { + case *plan.Literal_I8Val: + return strconv.FormatInt(int64(val.I8Val), 10), false, true + case *plan.Literal_I16Val: + return strconv.FormatInt(int64(val.I16Val), 10), false, true + case *plan.Literal_I32Val: + return strconv.FormatInt(int64(val.I32Val), 10), false, true + case *plan.Literal_I64Val: + return strconv.FormatInt(val.I64Val, 10), false, true + case *plan.Literal_U8Val: + return strconv.FormatUint(uint64(val.U8Val), 10), false, true + case *plan.Literal_U16Val: + return strconv.FormatUint(uint64(val.U16Val), 10), false, true + case *plan.Literal_U32Val: + return strconv.FormatUint(uint64(val.U32Val), 10), false, true + case *plan.Literal_U64Val: + return strconv.FormatUint(val.U64Val, 10), false, true + case *plan.Literal_Fval: + return strconv.FormatFloat(float64(val.Fval), 'g', -1, 32), false, true + case *plan.Literal_Dval: + return strconv.FormatFloat(val.Dval, 'g', -1, 64), false, true + case *plan.Literal_Bval: + return strconv.FormatBool(val.Bval), false, true + case *plan.Literal_EnumVal: + return strconv.FormatUint(uint64(val.EnumVal), 10), false, true + case *plan.Literal_Decimal64Val: + return types.Decimal64(val.Decimal64Val.A).Format(expr.Typ.Scale), false, true + case *plan.Literal_Decimal128Val: + decimal := types.Decimal128{ + B0_63: uint64(val.Decimal128Val.A), + B64_127: uint64(val.Decimal128Val.B), + } + return decimal.Format(expr.Typ.Scale), false, true + case *plan.Literal_Dateval: + return formatStringLiteral(types.Date(val.Dateval).String()), false, true + case *plan.Literal_Timeval: + return formatStringLiteral(types.Time(val.Timeval).String2(expr.Typ.Scale)), false, true + case *plan.Literal_Datetimeval: + return formatStringLiteral(types.Datetime(val.Datetimeval).String2(expr.Typ.Scale)), false, true + case *plan.Literal_Timestampval: + location := time.UTC + if proc := ctx.GetProcess(); proc != nil && proc.GetSessionInfo().TimeZone != nil { + location = proc.GetSessionInfo().TimeZone + } + value := types.Timestamp(val.Timestampval).String2(location, expr.Typ.Scale) + return formatStringLiteral(value), false, true + case *plan.Literal_Sval: + return formatStringLiteral(val.Sval), false, true + default: + return "", false, false + } + } + materializeDefault := func(colName string) (string, bool, error) { + col, found := findParentCol(colName) + if !found { + return "", false, moerr.NewInternalErrorf(ctx.GetContext(), + "REPLACE conflict column %s not found", colName) + } + if col.GeneratedCol != nil { + return "", false, moerr.NewNotSupported(ctx.GetContext(), + "REPLACE with an omitted generated conflict key") + } + defaultExpr, err := getDefaultExpr(ctx.GetContext(), col) + if err != nil { + return "", false, err + } + formatted, isNull, ok := formatLiteral(defaultExpr) + if !ok { + return "", false, moerr.NewNotSupported(ctx.GetContext(), + "REPLACE with a non-literal default conflict key") + } + return formatted, isNull, nil + } + sqlTypeForColumn := func(col *plan.ColDef) string { + typ := types.T(col.Typ.Id) + switch typ { + case types.T_time, types.T_datetime, types.T_timestamp: + if col.Typ.Scale > 0 { + return fmt.Sprintf("%s(%d)", makeTypeByPlan2Type(col.Typ).String(), col.Typ.Scale) + } + case types.T_enum: + values := strings.Split(col.Typ.Enumvalues, ",") + for i := range values { + values[i] = formatStringLiteral(values[i]) + } + return "ENUM(" + strings.Join(values, ",") + ")" + } + if isSetPlanType(&col.Typ) { + values := strings.Split(col.Typ.Enumvalues, ",") + for i := range values { + values[i] = formatStringLiteral(values[i]) + } + return "SET(" + strings.Join(values, ",") + ")" + } + return makeTypeByPlan2Type(col.Typ).DescString() + } + castForColumn := func(value string, col *plan.ColDef) string { + return fmt.Sprintf("cast(%s as %s)", value, sqlTypeForColumn(col)) + } + conflictPredicates := make([]string, 0, len(values.Rows)*len(uniqueKeys)) + for _, row := range values.Rows { + for _, key := range uniqueKeys { + parts := make([]string, 0, len(key.parts)) + keyCannotConflict := false + for _, colName := range key.parts { + col, found := findParentCol(colName) + if !found { + return "", nil, nil, moerr.NewInternalErrorf(ctx.GetContext(), + "REPLACE conflict column %s not found", colName) + } + pos, supplied := positions[colName] + var incomingExpr string + if !supplied || pos >= len(row) { + if col.Typ.AutoIncr { + keyCannotConflict = true + break + } + var isNull bool + var err error + incomingExpr, isNull, err = materializeDefault(colName) + if err != nil { + return "", nil, nil, err + } + if isNull { + keyCannotConflict = true + break + } + } else { + switch value := row[pos].(type) { + case *tree.NumVal: + if value.ValType == tree.P_char && + (types.T(col.Typ.Id) == types.T_char || types.T(col.Typ.Id) == types.T_varchar) && + col.Typ.Width > 0 && int32(utf8.RuneCountInString(value.String())) > col.Typ.Width { + return "", nil, nil, moerr.NewInvalidInputf(ctx.GetContext(), + "Src length %d is larger than Dest length %d", + utf8.RuneCountInString(value.String()), col.Typ.Width) + } + row[pos].Format(literalFmt) + incomingExpr = literalFmt.String() + literalFmt.Reset() + case *tree.StrVal: + if (types.T(col.Typ.Id) == types.T_char || types.T(col.Typ.Id) == types.T_varchar) && + col.Typ.Width > 0 && int32(utf8.RuneCountInString(value.String())) > col.Typ.Width { + return "", nil, nil, moerr.NewInvalidInputf(ctx.GetContext(), + "Src length %d is larger than Dest length %d", + utf8.RuneCountInString(value.String()), col.Typ.Width) + } + row[pos].Format(literalFmt) + incomingExpr = literalFmt.String() + literalFmt.Reset() + case *tree.DefaultVal: + var isNull bool + var err error + incomingExpr, isNull, err = materializeDefault(colName) + if err != nil { + return "", nil, nil, err + } + if isNull { + keyCannotConflict = true + break + } + default: + return "", nil, nil, moerr.NewNotSupported(ctx.GetContext(), "REPLACE with a non-literal conflict key") + } + } + if keyCannotConflict { + break + } + incomingExpr = castForColumn(incomingExpr, col) + parentExpr := qualifiedCol(parentAlias, colName) + if length := key.prefixLengths[colName]; length > 0 { + parentExpr = fmt.Sprintf("substring(%s, 1, %d)", parentExpr, length) + incomingExpr = fmt.Sprintf("substring(%s, 1, %d)", incomingExpr, length) + } + parts = append(parts, fmt.Sprintf("%s = %s", parentExpr, incomingExpr)) + } + if keyCannotConflict { + continue + } + conflictPredicates = append(conflictPredicates, "("+strings.Join(parts, " and ")+")") + } + } + if len(conflictPredicates) == 0 { + return "", nil, nil, nil + } + conflictPredicate := "(" + strings.Join(conflictPredicates, " or ") + ")" + parentTable := quoteIdentifier(parentRef.SchemaName) + "." + quoteIdentifier(parent.Name) + parentLock := fmt.Sprintf("select 1 from %s as %s where %s for update", + parentTable, quoteIdentifier(parentAlias), conflictPredicate) + + var checks, actions []string + seen := make(map[uint64]bool) + for _, childID := range parent.RefChildTbls { + if childID == 0 || seen[childID] { + continue + } + seen[childID] = true + childRef, child, err := ctx.ResolveById(childID, nil) + if err != nil { + return "", nil, nil, err + } + if child == nil { + return "", nil, nil, moerr.NewInternalError(ctx.GetContext(), fmt.Sprintf("referencing table %d not found", childID)) + } + for _, fk := range child.Fkeys { + if fk.ForeignTbl != parent.TblId { + continue + } + if len(fk.Cols) == 0 || len(fk.Cols) != len(fk.ForeignCols) { + return "", nil, nil, moerr.NewInternalError(ctx.GetContext(), "invalid parent foreign key definition") + } + parentCols, err := colIdsToNames(ctx.GetContext(), fk.ForeignCols, parent.Cols) + if err != nil { + return "", nil, nil, err + } + childCols, err := colIdsToNames(ctx.GetContext(), fk.Cols, child.Cols) + if err != nil { + return "", nil, nil, err + } + childTable := quoteIdentifier(childRef.SchemaName) + "." + quoteIdentifier(child.Name) + joinParts := make([]string, len(childCols)) + for i := range childCols { + joinParts[i] = fmt.Sprintf("%s.%s = %s", + childTable, quoteIdentifier(childCols[i]), qualifiedCol(parentAlias, parentCols[i])) + } + exists := fmt.Sprintf("exists (select 1 from %s as %s where %s and %s)", + parentTable, quoteIdentifier(parentAlias), conflictPredicate, strings.Join(joinParts, " and ")) + switch fk.OnDelete { + case plan.ForeignKeyDef_RESTRICT, plan.ForeignKeyDef_NO_ACTION, plan.ForeignKeyDef_SET_DEFAULT: + checks = append(checks, fmt.Sprintf("select count(*) = 0 from %s where %s", childTable, exists)) + case plan.ForeignKeyDef_CASCADE: + actions = append(actions, fmt.Sprintf("delete from %s where %s", childTable, exists)) + case plan.ForeignKeyDef_SET_NULL: + setParts := make([]string, len(childCols)) + for i, col := range childCols { + setParts[i] = quoteIdentifier(col) + " = null" + } + actions = append(actions, fmt.Sprintf("update %s set %s where %s", + childTable, strings.Join(setParts, ", "), exists)) + } + } + } + return parentLock, checks, actions, nil +} + func cleanHint(originSql string) string { re := regexp.MustCompile(`/\*[^!].*?\*/`) cleanSQL := re.ReplaceAllString(originSql, "") diff --git a/pkg/sql/plan/deepcopy.go b/pkg/sql/plan/deepcopy.go index 62e0d1a99bac8..9647a44745dd3 100644 --- a/pkg/sql/plan/deepcopy.go +++ b/pkg/sql/plan/deepcopy.go @@ -157,16 +157,21 @@ func DeepCopyLockTarget(target *plan.LockTarget) *plan.LockTarget { return nil } return &plan.LockTarget{ - TableId: target.TableId, - ObjRef: DeepCopyObjectRef(target.ObjRef), - PrimaryColIdxInBat: target.PrimaryColIdxInBat, - PrimaryColTyp: target.PrimaryColTyp, - RefreshTsIdxInBat: target.RefreshTsIdxInBat, - FilterColIdxInBat: target.FilterColIdxInBat, - LockTable: target.LockTable, - Block: target.Block, - LockRows: DeepCopyExpr(target.LockRows), - LockTableAtTheEnd: target.LockTableAtTheEnd, + TableId: target.TableId, + ObjRef: DeepCopyObjectRef(target.ObjRef), + PrimaryColIdxInBat: target.PrimaryColIdxInBat, + PrimaryColTyp: target.PrimaryColTyp, + RefreshTsIdxInBat: target.RefreshTsIdxInBat, + FilterColIdxInBat: target.FilterColIdxInBat, + LockTable: target.LockTable, + Block: target.Block, + Mode: target.Mode, + PrimaryColRelPos: target.PrimaryColRelPos, + FilterColRelPos: target.FilterColRelPos, + LockRows: DeepCopyExpr(target.LockRows), + LockTableAtTheEnd: target.LockTableAtTheEnd, + PartitionColIdxInBat: target.PartitionColIdxInBat, + HasPartitionCol: target.HasPartitionCol, } } @@ -591,6 +596,7 @@ func DeepCopyQuery(qry *plan.Query) *plan.Query { Params: DeepCopyExprList(qry.Params), Headings: qry.Headings, HasForeignKeyAction: qry.HasForeignKeyAction, + DetectSqls: slices.Clone(qry.DetectSqls), } for idx, node := range qry.Nodes { newQry.Nodes[idx] = DeepCopyNode(node) diff --git a/pkg/sql/plan/mock.go b/pkg/sql/plan/mock.go index 21f104cca5b22..92dd4d3a9d439 100644 --- a/pkg/sql/plan/mock.go +++ b/pkg/sql/plan/mock.go @@ -155,15 +155,16 @@ func NewEmptyCompilerContext() *MockCompilerContext { } type Schema struct { - cols []col - pks []int - idxs []index - fks []*ForeignKeyDef - clusterby *ClusterByDef - outcnt float64 - tblId int64 - isView bool - viewCfg ViewCfg + cols []col + pks []int + idxs []index + fks []*ForeignKeyDef + refChildTbls []uint64 + clusterby *ClusterByDef + outcnt float64 + tblId int64 + isView bool + viewCfg ViewCfg // tableType overrides TableType when non-empty; used to mock index tables // carrying an algo-specific type (e.g. ivfflat "metadata"). tableType string @@ -1040,6 +1041,173 @@ func NewMockCompilerContext(isDml bool) *MockCompilerContext { outcnt: 10, } + /* + Parent-side FK action fixtures for REPLACE (issue #24951). + + create table replace_fk_p(id int primary key, v varchar(20)); + create table replace_fk_c(id int primary key, pid int, + foreign key(pid) references replace_fk_p(id) on delete restrict); + + create table replace_fk_cp(id int primary key, v varchar(20)); + create table replace_fk_cc(id int primary key, pid int, + foreign key(pid) references replace_fk_cp(id) on delete cascade); + */ + constraintTestSchema["replace_fk_p"] = &Schema{ + tblId: 77001, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"v", types.T_varchar, true, 20, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + refChildTbls: []uint64{77002}, + outcnt: 4, + } + constraintTestSchema["replace_fk_c"] = &Schema{ + tblId: 77002, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"pid", types.T_int32, true, 32, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + fks: []*plan.ForeignKeyDef{ + { + Name: "fk_replace_c", + Cols: []uint64{1}, // pid + ForeignTbl: 77001, + ForeignCols: []uint64{0}, // replace_fk_p.id + OnDelete: plan.ForeignKeyDef_RESTRICT, + OnUpdate: plan.ForeignKeyDef_RESTRICT, + }, + }, + outcnt: 4, + } + constraintTestSchema["replace_fk_cp"] = &Schema{ + tblId: 77003, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"v", types.T_varchar, true, 20, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + refChildTbls: []uint64{77004}, + outcnt: 4, + } + constraintTestSchema["replace_fk_cc"] = &Schema{ + tblId: 77004, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"pid", types.T_int32, true, 32, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + fks: []*plan.ForeignKeyDef{ + { + Name: "fk_replace_cc", + Cols: []uint64{1}, // pid + ForeignTbl: 77003, + ForeignCols: []uint64{0}, // replace_fk_cp.id + OnDelete: plan.ForeignKeyDef_CASCADE, + OnUpdate: plan.ForeignKeyDef_CASCADE, + }, + }, + outcnt: 4, + } + constraintTestSchema["replace_fk_sp"] = &Schema{ + tblId: 77005, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"v", types.T_varchar, true, 20, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + refChildTbls: []uint64{77006}, + outcnt: 4, + } + constraintTestSchema["replace_fk_sc"] = &Schema{ + tblId: 77006, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"pid", types.T_int32, true, 32, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + fks: []*plan.ForeignKeyDef{ + { + Name: "fk_replace_sc", + Cols: []uint64{1}, // pid + ForeignTbl: 77005, + ForeignCols: []uint64{0}, // replace_fk_sp.id + OnDelete: plan.ForeignKeyDef_SET_NULL, + OnUpdate: plan.ForeignKeyDef_SET_NULL, + }, + }, + outcnt: 4, + } + constraintTestSchema["replace_fk_np"] = &Schema{ + tblId: 77007, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"v", types.T_varchar, true, 20, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + refChildTbls: []uint64{77008}, + outcnt: 4, + } + constraintTestSchema["replace_fk_nc"] = &Schema{ + tblId: 77008, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"pid", types.T_int32, true, 32, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + fks: []*plan.ForeignKeyDef{ + { + Name: "fk_replace_nc", + Cols: []uint64{1}, // pid + ForeignTbl: 77007, + ForeignCols: []uint64{0}, // replace_fk_np.id + OnDelete: plan.ForeignKeyDef_NO_ACTION, + OnUpdate: plan.ForeignKeyDef_NO_ACTION, + }, + }, + outcnt: 4, + } + constraintTestSchema["replace_fk_dp"] = &Schema{ + tblId: 77009, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"v", types.T_varchar, true, 20, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + refChildTbls: []uint64{77010}, + outcnt: 4, + } + constraintTestSchema["replace_fk_dc"] = &Schema{ + tblId: 77010, + cols: []col{ + {"id", types.T_int32, true, 32, 0}, + {"pid", types.T_int32, true, 32, 0}, + {catalog.Row_ID, types.T_Rowid, false, 16, 0}, + }, + pks: []int{0}, + fks: []*plan.ForeignKeyDef{ + { + Name: "fk_replace_dc", + Cols: []uint64{1}, // pid + ForeignTbl: 77009, + ForeignCols: []uint64{0}, // replace_fk_dp.id + OnDelete: plan.ForeignKeyDef_SET_DEFAULT, + OnUpdate: plan.ForeignKeyDef_SET_DEFAULT, + }, + }, + outcnt: 4, + } + /* create table products ( pid int not null, @@ -1513,6 +1681,10 @@ func NewMockCompilerContext(isDml bool) *MockCompilerContext { tableDef.Fkeys = table.fks } + if table.refChildTbls != nil { + tableDef.RefChildTbls = table.refChildTbls + } + if table.clusterby != nil { tableDef.ClusterBy = &plan.ClusterByDef{ Name: "__mo_cbkey_003pid005pname", diff --git a/pkg/sql/plan/utils_test.go b/pkg/sql/plan/utils_test.go index dc10cc0c1ff84..26b840b8c4719 100644 --- a/pkg/sql/plan/utils_test.go +++ b/pkg/sql/plan/utils_test.go @@ -144,7 +144,8 @@ func TestFillValuesOfParamsInPlanDoesNotMutatePreparedPlan(t *testing.T) { }}} queryPlan := &plan.Plan{ Plan: &plan.Plan_Query{Query: &plan.Query{ - Steps: []int32{0}, + Steps: []int32{0}, + DetectSqls: []string{"REPLACE_PARENT_LOCK:select 1 for update", "REPLACE_PARENT_CHK:select true"}, Nodes: []*plan.Node{{ NodeType: plan.Node_VALUE_SCAN, Limit: &plan.Expr{Expr: &plan.Expr_P{P: &plan.ParamRef{Pos: 0}}}, @@ -177,6 +178,9 @@ func TestFillValuesOfParamsInPlanDoesNotMutatePreparedPlan(t *testing.T) { require.Equal(t, "AB\x00\x00", copiedLiteral.GetSval()) require.NotSame(t, source, copiedLiteral.GetSrc()) require.NotNil(t, binaryLiteral.GetLit().GetSrc().GetP()) + require.Equal(t, queryPlan.GetQuery().DetectSqls, filled.GetQuery().DetectSqls) + filled.GetQuery().DetectSqls[0] = "changed" + require.Equal(t, "REPLACE_PARENT_LOCK:select 1 for update", queryPlan.GetQuery().DetectSqls[0]) } } diff --git a/test/distributed/cases/dml/replace/replace.result b/test/distributed/cases/dml/replace/replace.result index f8651c6949111..f53b157a4a440 100644 --- a/test/distributed/cases/dml/replace/replace.result +++ b/test/distributed/cases/dml/replace/replace.result @@ -287,6 +287,274 @@ name email value a y@a.com 111 b z@a.com 222 drop table t_replace_multi_uk_batch; +drop table if exists fk_c; +drop table if exists fk_p; +create table fk_p(id int primary key, v varchar(20)); +create table fk_c(id int primary key, pid int, foreign key(pid) references fk_p(id) on delete restrict); +insert into fk_p values (1,'p1'); +insert into fk_c values (10,1); +replace into fk_p values (1,'p1_new'); +Cannot delete or update a parent row: a foreign key constraint fails +select * from fk_p order by id; +id v +1 p1 +select * from fk_c order by id; +id pid +10 1 +replace into fk_p values (2,'p2_new'); +select * from fk_p order by id; +id v +1 p1 +2 p2_new +drop table fk_c; +drop table fk_p; +drop table if exists fk_cc; +drop table if exists fk_cp; +create table fk_cp(id int primary key, v varchar(20)); +create table fk_cc(id int primary key, pid int, foreign key(pid) references fk_cp(id) on delete cascade); +insert into fk_cp values (1,'p1'); +insert into fk_cc values (10,1); +replace into fk_cp values (1,'p1_new'); +select * from fk_cp order by id; +id v +1 p1_new +select * from fk_cc order by id; +id pid +drop table fk_cc; +drop table fk_cp; +drop table if exists fk_sc; +drop table if exists fk_sp; +create table fk_sp(id int primary key, v varchar(20)); +create table fk_sc(id int primary key, pid int, foreign key(pid) references fk_sp(id) on delete set null); +insert into fk_sp values (1,'p1'); +insert into fk_sc values (10,1); +replace into fk_sp values (1,'p1_new'); +select * from fk_sp order by id; +id v +1 p1_new +select * from fk_sc order by id; +id pid +10 null +drop table fk_sc; +drop table fk_sp; +drop table if exists fk_dc; +drop table if exists fk_dp; +create table fk_dp(id int primary key, v varchar(20)); +create table fk_dc(id int primary key, pid int default 2, +foreign key(pid) references fk_dp(id) on delete set default); +insert into fk_dp values (1,'p1'),(2,'p2'); +insert into fk_dc values (10,1); +replace into fk_dp values (1,'p1_new'); +Cannot delete or update a parent row: a foreign key constraint fails +select * from fk_dp order by id; +id v +1 p1 +2 p2 +select * from fk_dc order by id; +id pid +10 1 +drop table fk_dc; +drop table fk_dp; +drop table if exists fk_mc; +drop table if exists fk_mp; +create table fk_mp(id int primary key, v varchar(20)); +create table fk_mc(id int primary key, pid int, foreign key(pid) references fk_mp(id) on delete cascade); +insert into fk_mp values (1,'p1'),(2,'p2'),(3,'p3'); +insert into fk_mc values (10,1),(20,2),(30,3); +replace into fk_mp values (1,'p1_new'),(2,'p2_new'); +select * from fk_mp order by id; +id v +1 p1_new +2 p2_new +3 p3 +select * from fk_mc order by id; +id pid +30 3 +drop table fk_mc; +drop table fk_mp; +drop table if exists fk_review_c; +drop table if exists fk_review_p; +create table fk_review_p(id int primary key, u int unique, v int); +create table fk_review_c(id int primary key, pid int, +foreign key(pid) references fk_review_p(id) on delete cascade); +insert into fk_review_p values (1, 10, 100); +insert into fk_review_c values (1, 1); +replace into fk_review_p values (2, 10, 200); +select * from fk_review_p order by id; +id u v +2 10 200 +select * from fk_review_c order by id; +id pid +create table fk_review_src(id int, u int, v int); +insert into fk_review_src values (1, 10, 400); +replace into fk_review_p select * from fk_review_src; +not supported: REPLACE SELECT/TABLE on a referenced parent table +select * from fk_review_p order by id; +id u v +2 10 200 +select * from fk_review_c order by id; +id pid +drop table fk_review_src; +drop table fk_review_c; +drop table fk_review_p; +create table fk_auto_p(id int auto_increment primary key, u int unique, v int); +create table fk_auto_c(id int primary key, pid int, +foreign key(pid) references fk_auto_p(id) on delete cascade); +insert into fk_auto_p(u, v) values (10, 100); +insert into fk_auto_c values (1, 1); +replace into fk_auto_p(u, v) values (10, 200); +select u, v from fk_auto_p; +u v +10 200 +select * from fk_auto_c; +id pid +drop table fk_auto_c; +drop table fk_auto_p; +create table fk_nonpk_p(id int primary key, u int unique); +create table fk_nonpk_c(id int primary key, parent_u int, +foreign key(parent_u) references fk_nonpk_p(u) on delete cascade); +insert into fk_nonpk_p values (1, 10); +insert into fk_nonpk_c values (1, 10); +replace into fk_nonpk_p values (2, 10); +select * from fk_nonpk_p; +id u +2 10 +select * from fk_nonpk_c; +id parent_u +drop table fk_nonpk_c; +drop table fk_nonpk_p; +create table fk_fanout_p(id int primary key, u int unique, v int unique); +create table fk_fanout_c(id int primary key, pid int, +foreign key(pid) references fk_fanout_p(id) on delete cascade); +insert into fk_fanout_p values (1, 10, 100), (2, 20, 200); +insert into fk_fanout_c values (1, 1), (2, 2); +replace into fk_fanout_p values (3, 10, 200); +select * from fk_fanout_p; +id u v +3 10 200 +select * from fk_fanout_c; +id pid +drop table fk_fanout_c; +drop table fk_fanout_p; +create table fk_prefix_p(id int primary key, body varchar(64), unique key u(body(4))); +create table fk_prefix_c(id int primary key, pid int, +foreign key(pid) references fk_prefix_p(id) on delete cascade); +insert into fk_prefix_p values (1, 'abcdxxxx'); +insert into fk_prefix_c values (1, 1); +replace into fk_prefix_p values (2, 'abcdyyyy'); +select * from fk_prefix_p; +id body +2 abcdyyyy +select * from fk_prefix_c; +id pid +drop table fk_prefix_c; +drop table fk_prefix_p; +create table fk_omitted_uk_p( +id int primary key, +u varchar(20) unique, +v int +); +create table fk_omitted_uk_c( +id int primary key, +pid int, +foreign key(pid) references fk_omitted_uk_p(id) on delete cascade +); +insert into fk_omitted_uk_p values (1, 'x', 100); +insert into fk_omitted_uk_c values (1, 1); +replace into fk_omitted_uk_p(id) values (1); +select * from fk_omitted_uk_p; +id u v +1 NULL NULL +select * from fk_omitted_uk_c; +id pid +drop table fk_omitted_uk_c; +drop table fk_omitted_uk_p; +create table fk_temporal_default_p( +id int primary key, +d date unique default '2026-07-15', +t time unique default '12:34:56', +dt datetime unique default '2026-07-15 12:34:56', +ts timestamp unique default '2026-07-15 12:34:56' +); +create table fk_temporal_default_c( +id int primary key, +pid int, +foreign key(pid) references fk_temporal_default_p(id) on delete cascade +); +insert into fk_temporal_default_p(id) values (1); +insert into fk_temporal_default_c values (1, 1); +replace into fk_temporal_default_p(id) values (2); +select * from fk_temporal_default_p; +id d t dt ts +2 2026-07-15 12:34:56 2026-07-15 12:34:56 2026-07-15 12:34:56 +select * from fk_temporal_default_c; +id pid +drop table fk_temporal_default_c; +drop table fk_temporal_default_p; +create table fk_decimal_cast_p(id int primary key, u decimal(5,2) unique); +create table fk_decimal_cast_c(id int primary key, pid int, +foreign key(pid) references fk_decimal_cast_p(id) on delete cascade); +insert into fk_decimal_cast_p values (1, 1.23); +insert into fk_decimal_cast_c values (1, 1); +replace into fk_decimal_cast_p values (2, 1.234); +select * from fk_decimal_cast_p; +id u +2 1.23 +select * from fk_decimal_cast_c; +id pid +drop table fk_decimal_cast_c; +drop table fk_decimal_cast_p; +create table fk_param_p(id int primary key, v int); +create table fk_param_c(id int primary key, pid int, +foreign key(pid) references fk_param_p(id) on delete restrict); +prepare fk_param_key_stmt from 'replace into fk_param_p values (?, ?)'; +not supported: REPLACE with a non-literal conflict key +set @fk_id = 1, @fk_v = 300; +execute fk_param_key_stmt using @fk_id, @fk_v; +invalid state prepared statement 'fk_param_key_stmt' does not exist +deallocate prepare fk_param_key_stmt; +insert into fk_param_p values (1, 100); +insert into fk_param_c values (1, 1); +prepare fk_review_stmt from 'replace into fk_param_p values (1, ?)'; +set @fk_v = 300; +execute fk_review_stmt using @fk_v; +Cannot delete or update a parent row: a foreign key constraint fails +select * from fk_param_p; +id v +1 100 +select * from fk_param_c; +id pid +1 1 +deallocate prepare fk_review_stmt; +drop table fk_param_c; +drop table fk_param_p; +drop table if exists fk_comp_c; +drop table if exists fk_comp_p; +create table fk_comp_p(a int, b int, primary key(a, b)); +create table fk_comp_c(id int primary key, a int, b int, +foreign key(a, b) references fk_comp_p(a, b) on delete cascade); +insert into fk_comp_p values (1, 1); +insert into fk_comp_c values (1, 1, 1); +replace into fk_comp_p values (1, 1); +select * from fk_comp_p; +a b +1 1 +select * from fk_comp_c; +id a b +drop table fk_comp_c; +drop table fk_comp_p; +drop table if exists `fk``tick_c`; +drop table if exists `fk``tick_p`; +create table `fk``tick_p`(`id``x` int primary key); +create table `fk``tick_c`(id int primary key, `pid``x` int, +foreign key(`pid``x`) references `fk``tick_p`(`id``x`) on delete cascade); +insert into `fk``tick_p` values (1); +insert into `fk``tick_c` values (1, 1); +replace into `fk``tick_p` values (1); +select * from `fk``tick_c`; +id pid`x +drop table `fk``tick_c`; +drop table `fk``tick_p`; drop table if exists replace_fk_c; drop table if exists replace_fk_p; create table replace_fk_p(id int primary key); diff --git a/test/distributed/cases/dml/replace/replace.test b/test/distributed/cases/dml/replace/replace.test index bb46ebaa75672..e10cb85daaa7e 100644 --- a/test/distributed/cases/dml/replace/replace.test +++ b/test/distributed/cases/dml/replace/replace.test @@ -237,6 +237,232 @@ insert into t_replace_multi_uk_batch (name, email, value) values ('a', 'x@a.com' replace into t_replace_multi_uk_batch (name, email, value) values ('a', 'y@a.com', 111), ('b', 'z@a.com', 222); select name, email, value from t_replace_multi_uk_batch order by name; drop table t_replace_multi_uk_batch; +-- parent-side foreign key actions during REPLACE (delete-then-insert semantics) +-- ON DELETE RESTRICT: replacing a referenced parent row must fail and leave it unchanged +drop table if exists fk_c; +drop table if exists fk_p; +create table fk_p(id int primary key, v varchar(20)); +create table fk_c(id int primary key, pid int, foreign key(pid) references fk_p(id) on delete restrict); +insert into fk_p values (1,'p1'); +insert into fk_c values (10,1); +replace into fk_p values (1,'p1_new'); +select * from fk_p order by id; +select * from fk_c order by id; +-- replacing an unreferenced parent row is allowed +replace into fk_p values (2,'p2_new'); +select * from fk_p order by id; +drop table fk_c; +drop table fk_p; + +-- ON DELETE CASCADE: replacing a referenced parent row cascades the delete to children +drop table if exists fk_cc; +drop table if exists fk_cp; +create table fk_cp(id int primary key, v varchar(20)); +create table fk_cc(id int primary key, pid int, foreign key(pid) references fk_cp(id) on delete cascade); +insert into fk_cp values (1,'p1'); +insert into fk_cc values (10,1); +replace into fk_cp values (1,'p1_new'); +select * from fk_cp order by id; +select * from fk_cc order by id; +drop table fk_cc; +drop table fk_cp; + +-- ON DELETE SET NULL: replacing a referenced parent row nulls the child fk columns +drop table if exists fk_sc; +drop table if exists fk_sp; +create table fk_sp(id int primary key, v varchar(20)); +create table fk_sc(id int primary key, pid int, foreign key(pid) references fk_sp(id) on delete set null); +insert into fk_sp values (1,'p1'); +insert into fk_sc values (10,1); +replace into fk_sp values (1,'p1_new'); +select * from fk_sp order by id; +select * from fk_sc order by id; +drop table fk_sc; +drop table fk_sp; + +-- ON DELETE SET DEFAULT follows DELETE semantics and restricts referenced parents +drop table if exists fk_dc; +drop table if exists fk_dp; +create table fk_dp(id int primary key, v varchar(20)); +create table fk_dc(id int primary key, pid int default 2, + foreign key(pid) references fk_dp(id) on delete set default); +insert into fk_dp values (1,'p1'),(2,'p2'); +insert into fk_dc values (10,1); +replace into fk_dp values (1,'p1_new'); +select * from fk_dp order by id; +select * from fk_dc order by id; +drop table fk_dc; +drop table fk_dp; + +-- multi-row REPLACE: every conflicting parent row applies its child action (CASCADE) +drop table if exists fk_mc; +drop table if exists fk_mp; +create table fk_mp(id int primary key, v varchar(20)); +create table fk_mc(id int primary key, pid int, foreign key(pid) references fk_mp(id) on delete cascade); +insert into fk_mp values (1,'p1'),(2,'p2'),(3,'p3'); +insert into fk_mc values (10,1),(20,2),(30,3); +replace into fk_mp values (1,'p1_new'),(2,'p2_new'); +select * from fk_mp order by id; +select * from fk_mc order by id; +drop table fk_mc; +drop table fk_mp; + +-- Unsupported parent-side FK shapes must fail instead of silently skipping actions. +drop table if exists fk_review_c; +drop table if exists fk_review_p; +create table fk_review_p(id int primary key, u int unique, v int); +create table fk_review_c(id int primary key, pid int, + foreign key(pid) references fk_review_p(id) on delete cascade); +insert into fk_review_p values (1, 10, 100); +insert into fk_review_c values (1, 1); +replace into fk_review_p values (2, 10, 200); +select * from fk_review_p order by id; +select * from fk_review_c order by id; +create table fk_review_src(id int, u int, v int); +insert into fk_review_src values (1, 10, 400); +replace into fk_review_p select * from fk_review_src; +select * from fk_review_p order by id; +select * from fk_review_c order by id; +drop table fk_review_src; +drop table fk_review_c; +drop table fk_review_p; + +create table fk_auto_p(id int auto_increment primary key, u int unique, v int); +create table fk_auto_c(id int primary key, pid int, + foreign key(pid) references fk_auto_p(id) on delete cascade); +insert into fk_auto_p(u, v) values (10, 100); +insert into fk_auto_c values (1, 1); +replace into fk_auto_p(u, v) values (10, 200); +select u, v from fk_auto_p; +select * from fk_auto_c; +drop table fk_auto_c; +drop table fk_auto_p; + +create table fk_nonpk_p(id int primary key, u int unique); +create table fk_nonpk_c(id int primary key, parent_u int, + foreign key(parent_u) references fk_nonpk_p(u) on delete cascade); +insert into fk_nonpk_p values (1, 10); +insert into fk_nonpk_c values (1, 10); +replace into fk_nonpk_p values (2, 10); +select * from fk_nonpk_p; +select * from fk_nonpk_c; +drop table fk_nonpk_c; +drop table fk_nonpk_p; + +create table fk_fanout_p(id int primary key, u int unique, v int unique); +create table fk_fanout_c(id int primary key, pid int, + foreign key(pid) references fk_fanout_p(id) on delete cascade); +insert into fk_fanout_p values (1, 10, 100), (2, 20, 200); +insert into fk_fanout_c values (1, 1), (2, 2); +replace into fk_fanout_p values (3, 10, 200); +select * from fk_fanout_p; +select * from fk_fanout_c; +drop table fk_fanout_c; +drop table fk_fanout_p; + +create table fk_prefix_p(id int primary key, body varchar(64), unique key u(body(4))); +create table fk_prefix_c(id int primary key, pid int, + foreign key(pid) references fk_prefix_p(id) on delete cascade); +insert into fk_prefix_p values (1, 'abcdxxxx'); +insert into fk_prefix_c values (1, 1); +replace into fk_prefix_p values (2, 'abcdyyyy'); +select * from fk_prefix_p; +select * from fk_prefix_c; +drop table fk_prefix_c; +drop table fk_prefix_p; + +create table fk_omitted_uk_p( + id int primary key, + u varchar(20) unique, + v int +); +create table fk_omitted_uk_c( + id int primary key, + pid int, + foreign key(pid) references fk_omitted_uk_p(id) on delete cascade +); +insert into fk_omitted_uk_p values (1, 'x', 100); +insert into fk_omitted_uk_c values (1, 1); +replace into fk_omitted_uk_p(id) values (1); +select * from fk_omitted_uk_p; +select * from fk_omitted_uk_c; +drop table fk_omitted_uk_c; +drop table fk_omitted_uk_p; + +create table fk_temporal_default_p( + id int primary key, + d date unique default '2026-07-15', + t time unique default '12:34:56', + dt datetime unique default '2026-07-15 12:34:56', + ts timestamp unique default '2026-07-15 12:34:56' +); +create table fk_temporal_default_c( + id int primary key, + pid int, + foreign key(pid) references fk_temporal_default_p(id) on delete cascade +); +insert into fk_temporal_default_p(id) values (1); +insert into fk_temporal_default_c values (1, 1); +replace into fk_temporal_default_p(id) values (2); +select * from fk_temporal_default_p; +select * from fk_temporal_default_c; +drop table fk_temporal_default_c; +drop table fk_temporal_default_p; + +create table fk_decimal_cast_p(id int primary key, u decimal(5,2) unique); +create table fk_decimal_cast_c(id int primary key, pid int, + foreign key(pid) references fk_decimal_cast_p(id) on delete cascade); +insert into fk_decimal_cast_p values (1, 1.23); +insert into fk_decimal_cast_c values (1, 1); +replace into fk_decimal_cast_p values (2, 1.234); +select * from fk_decimal_cast_p; +select * from fk_decimal_cast_c; +drop table fk_decimal_cast_c; +drop table fk_decimal_cast_p; + +create table fk_param_p(id int primary key, v int); +create table fk_param_c(id int primary key, pid int, + foreign key(pid) references fk_param_p(id) on delete restrict); +prepare fk_param_key_stmt from 'replace into fk_param_p values (?, ?)'; +set @fk_id = 1, @fk_v = 300; +execute fk_param_key_stmt using @fk_id, @fk_v; +deallocate prepare fk_param_key_stmt; +insert into fk_param_p values (1, 100); +insert into fk_param_c values (1, 1); +prepare fk_review_stmt from 'replace into fk_param_p values (1, ?)'; +set @fk_v = 300; +execute fk_review_stmt using @fk_v; +select * from fk_param_p; +select * from fk_param_c; +deallocate prepare fk_review_stmt; +drop table fk_param_c; +drop table fk_param_p; + +drop table if exists fk_comp_c; +drop table if exists fk_comp_p; +create table fk_comp_p(a int, b int, primary key(a, b)); +create table fk_comp_c(id int primary key, a int, b int, + foreign key(a, b) references fk_comp_p(a, b) on delete cascade); +insert into fk_comp_p values (1, 1); +insert into fk_comp_c values (1, 1, 1); +replace into fk_comp_p values (1, 1); +select * from fk_comp_p; +select * from fk_comp_c; +drop table fk_comp_c; +drop table fk_comp_p; + +drop table if exists `fk``tick_c`; +drop table if exists `fk``tick_p`; +create table `fk``tick_p`(`id``x` int primary key); +create table `fk``tick_c`(id int primary key, `pid``x` int, + foreign key(`pid``x`) references `fk``tick_p`(`id``x`) on delete cascade); +insert into `fk``tick_p` values (1); +insert into `fk``tick_c` values (1, 1); +replace into `fk``tick_p` values (1); +select * from `fk``tick_c`; +drop table `fk``tick_c`; +drop table `fk``tick_p`; + -- child-side foreign key check during REPLACE: an inserted/replaced child row -- must reference an existing parent row, otherwise REPLACE fails and the -- previous conflicting row stays intact