fix(plan): dedup update from source matches

pkg/sql/colexec/aggexec/any2.go

@@ -48,7 +48,9 @@ func (exec *anyExec) BatchFill(offset int, groups []uint64, vectors []*vector.Ve
 			if exec.state[x].vecs[0].IsNull(uint64(y)) {
 				exec.state[x].vecs[0].UnsetNull(uint64(y))
 				bs := vectors[0].GetRawBytesAt(int(idx))
-				exec.state[x].vecs[0].SetRawBytesAt(int(y), bs, exec.mp)
+				if err := exec.state[x].vecs[0].SetRawBytesAt(int(y), bs, exec.mp); err != nil {
+					return err
+				}
 			}
 		}
 	}
@@ -74,7 +76,9 @@ func (exec *anyExec) BatchMerge(next AggFuncExec, offset int, groups []uint64) e
 		if exec.state[x1].vecs[0].IsNull(uint64(y1)) {
 			exec.state[x1].vecs[0].UnsetNull(uint64(y1))
 			bs := other.state[x2].vecs[0].GetRawBytesAt(int(y2))
-			exec.state[x1].vecs[0].SetRawBytesAt(int(y1), bs, exec.mp)
+			if err := exec.state[x1].vecs[0].SetRawBytesAt(int(y1), bs, exec.mp); err != nil {
+				return err
+			}
 		}
 	}
 	return nil

pkg/sql/colexec/aggexec/any2_test.go

@@ -0,0 +1,81 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package aggexec
+
+import (
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAnyValueBatchFillReturnsSetRawBytesAtError(t *testing.T) {
+	inputMp := mpool.MustNewZero()
+	input := vector.NewVec(types.T_varchar.ToType())
+	require.NoError(t, vector.AppendBytes(input, make([]byte, 4096), false, inputMp))
+	defer input.Free(inputMp)
+
+	exec, limitedMp, filler := newLimitedAnyValueExec(t)
+	defer cleanupLimitedAnyValueExec(exec, limitedMp, filler)
+
+	err := exec.BatchFill(0, []uint64{1}, []*vector.Vector{input})
+	require.Error(t, err)
+}
+
+func TestAnyValueBatchMergeReturnsSetRawBytesAtError(t *testing.T) {
+	inputMp := mpool.MustNewZero()
+	input := vector.NewVec(types.T_varchar.ToType())
+	require.NoError(t, vector.AppendBytes(input, make([]byte, 4096), false, inputMp))
+	defer input.Free(inputMp)
+
+	sourceMp := mpool.MustNewZero()
+	source := makeAnyValueExec(sourceMp, 1, types.T_varchar.ToType()).(*anyExec)
+	require.NoError(t, source.GroupGrow(1))
+	require.NoError(t, source.BatchFill(0, []uint64{1}, []*vector.Vector{input}))
+	defer source.Free()
+
+	target, limitedMp, filler := newLimitedAnyValueExec(t)
+	defer cleanupLimitedAnyValueExec(target, limitedMp, filler)
+
+	err := target.BatchMerge(source, 0, []uint64{1})
+	require.Error(t, err)
+}
+
+func newLimitedAnyValueExec(t *testing.T) (*anyExec, *mpool.MPool, []byte) {
+	t.Helper()
+
+	limitedMp, err := mpool.NewMPool("any-value-limited", 1024*1024, mpool.NoFixed)
+	require.NoError(t, err)
+
+	exec := makeAnyValueExec(limitedMp, 1, types.T_varchar.ToType()).(*anyExec)
+	require.NoError(t, exec.GroupGrow(1))
+
+	remaining := 1024*1024 - limitedMp.CurrNB()
+	require.Greater(t, remaining, int64(4096))
+	filler, err := limitedMp.Alloc(int(remaining-1024), true)
+	require.NoError(t, err)
+
+	return exec, limitedMp, filler
+}
+
+func cleanupLimitedAnyValueExec(exec *anyExec, mp *mpool.MPool, filler []byte) {
+	if filler != nil {
+		mp.Free(filler)
+	}
+	exec.Free()
+	mpool.DeleteMPool(mp)
+}

pkg/sql/plan/bind_update.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 )
@@ -269,7 +270,25 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext)
 			}
 		}
 
-		// Recompute generated columns (both STORED and VIRTUAL are computed on write)
+	}
+
+	if stmt.From != nil && len(stmt.From.Tables) > 0 {
+		lastNodeID, selectNode, selectNodeTag, err = builder.appendUpdateFromDedupNode(
+			bindCtx, lastNodeID, selectNode, selectNodeTag, dmlCtx, oldColName2Idx, newColName2Idx)
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	for i, alias := range dmlCtx.aliases {
+		if len(dmlCtx.updateCol2Expr[i]) == 0 {
+			continue
+		}
+
+		tableDef := dmlCtx.tableDefs[i]
+
+		// Recompute generated columns after UPDATE FROM dedup so generated
+		// expressions read the same deduped base values that will be written.
 		for _, col := range tableDef.Cols {
 			if col.GeneratedCol == nil {
 				continue
@@ -957,3 +976,175 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext)
 
 	return lastNodeID, err
 }
+
+func (builder *QueryBuilder) appendUpdateFromDedupNode(
+	bindCtx *BindContext,
+	lastNodeID int32,
+	selectNode *plan.Node,
+	selectNodeTag int32,
+	dmlCtx *DMLContext,
+	oldColName2Idx map[string]int32,
+	newColName2Idx map[string]int32,
+) (int32, *plan.Node, int32, error) {
+	partitionByExprs := make([]*plan.Expr, 0)
+	partitionPos := make(map[int32]struct{})
+	childColExpr := func(pos int32) *plan.Expr {
+		e := selectNode.ProjectList[pos]
+		name := ""
+		if col, ok := e.Expr.(*plan.Expr_Col); ok {
+			name = col.Col.Name
+		}
+		return &plan.Expr{
+			Typ: e.Typ,
+			Expr: &plan.Expr_Col{
+				Col: &plan.ColRef{
+					RelPos: selectNodeTag,
+					ColPos: pos,
+					Name:   name,
+				},
+			},
+		}
+	}
+
+	for i, alias := range dmlCtx.aliases {
+		if len(dmlCtx.updateCol2Expr[i]) == 0 {
+			continue
+		}
+
+		for _, col := range dmlCtx.tableDefs[i].Cols {
+			key := alias + "." + col.Name
+			oldPos, ok := oldColName2Idx[key]
+			if !ok {
+				continue
+			}
+			if _, exists := partitionPos[oldPos]; !exists {
+				partitionPos[oldPos] = struct{}{}
+				partitionByExprs = append(partitionByExprs, childColExpr(oldPos))
+			}
+		}
+	}
+
+	if len(partitionByExprs) == 0 {
+		return lastNodeID, selectNode, selectNodeTag, nil
+	}
+
+	windowTag := builder.genNewBindTag()
+	partitionBy := make([]*plan.OrderBySpec, 0, len(partitionByExprs))
+	for _, expr := range partitionByExprs {
+		partitionBy = append(partitionBy, &plan.OrderBySpec{
+			Expr: expr,
+			Flag: plan.OrderBySpec_INTERNAL,
+		})
+	}
+	lastNodeID = builder.appendNode(&plan.Node{
+		NodeType:    plan.Node_PARTITION,
+		Children:    []int32{lastNodeID},
+		OrderBy:     partitionBy,
+		BindingTags: []int32{windowTag},
+	}, bindCtx)
+
+	rowNumberFunc, err := BindFuncExprImplByPlanExpr(builder.GetContext(), "row_number", nil)
+	if err != nil {
+		return 0, nil, 0, err
+	}
+	rowNumberExpr := &plan.Expr{
+		Typ: rowNumberFunc.Typ,
+		Expr: &plan.Expr_W{
+			W: &plan.WindowSpec{
+				WindowFunc:  rowNumberFunc,
+				Name:        "row_number",
+				PartitionBy: partitionByExprs,
+				Frame: &plan.FrameClause{
+					Type: plan.FrameClause_ROWS,
+					Start: &plan.FrameBound{
+						Type:      plan.FrameBound_PRECEDING,
+						UnBounded: true,
+					},
+					End: &plan.FrameBound{
+						Type:      plan.FrameBound_FOLLOWING,
+						UnBounded: true,
+					},
+				},
+			},
+		},
+	}
+	rowNumberIdx := int32(0)
+	rowNumberProjectPos := int32(len(selectNode.ProjectList))
+	lastNodeID = builder.appendNode(&plan.Node{
+		NodeType:    plan.Node_WINDOW,
+		Children:    []int32{lastNodeID},
+		WinSpecList: []*plan.Expr{rowNumberExpr},
+		WindowIdx:   rowNumberIdx,
+		BindingTags: []int32{windowTag},
+	}, bindCtx)
+
+	windowProjectTag := builder.genNewBindTag()
+	windowProjectList := make([]*plan.Expr, 0, len(selectNode.ProjectList)+1)
+	for pos := range selectNode.ProjectList {
+		windowProjectList = append(windowProjectList, childColExpr(int32(pos)))
+	}
+	windowProjectList = append(windowProjectList, &plan.Expr{
+		Typ: rowNumberFunc.Typ,
+		Expr: &plan.Expr_Col{
+			Col: &plan.ColRef{
+				RelPos: windowTag,
+				ColPos: rowNumberIdx,
+				Name:   "__mo_update_from_dedup_row_number",
+			},
+		},
+	})
+	lastNodeID = builder.appendNode(&plan.Node{
+		NodeType:    plan.Node_PROJECT,
+		Children:    []int32{lastNodeID},
+		ProjectList: windowProjectList,
+		BindingTags: []int32{windowProjectTag},
+	}, bindCtx)
+
+	rowNumberCol := &plan.Expr{
+		Typ: plan.Type{Id: int32(types.T_int64), NotNullable: true},
+		Expr: &plan.Expr_Col{
+			Col: &plan.ColRef{
+				RelPos: windowProjectTag,
+				ColPos: rowNumberProjectPos,
+				Name:   "__mo_update_from_dedup_row_number",
+			},
+		},
+	}
+	keepFirstRowExpr, err := BindFuncExprImplByPlanExpr(
+		builder.GetContext(), "=", []*plan.Expr{rowNumberCol, makePlan2Int64ConstExprWithType(1)})
+	if err != nil {
+		return 0, nil, 0, err
+	}
+	lastNodeID = builder.appendNode(&plan.Node{
+		NodeType:   plan.Node_FILTER,
+		Children:   []int32{lastNodeID},
+		FilterList: []*plan.Expr{keepFirstRowExpr},
+	}, bindCtx)
+
+	projectList := make([]*plan.Expr, len(selectNode.ProjectList))
+	for pos, e := range selectNode.ProjectList {
+		name := ""
+		if col, ok := e.Expr.(*plan.Expr_Col); ok {
+			name = col.Col.Name
+		}
+		projectList[pos] = &plan.Expr{
+			Typ: e.Typ,
+			Expr: &plan.Expr_Col{
+				Col: &plan.ColRef{
+					RelPos: windowProjectTag,
+					ColPos: int32(pos),
+					Name:   name,
+				},
+			},
+		}
+	}
+
+	projectNode := &plan.Node{
+		NodeType:    plan.Node_PROJECT,
+		Children:    []int32{lastNodeID},
+		ProjectList: projectList,
+		BindingTags: []int32{builder.genNewBindTag()},
+	}
+	lastNodeID = builder.appendNode(projectNode, bindCtx)
+	return lastNodeID, projectNode, projectNode.BindingTags[0], nil
+}