fix(sql): preserve signed zero in float set operations

pkg/common/bitmap/bitmap.go

@@ -266,6 +266,15 @@ func (n *Bitmap) ReleaseExternalStorage() []uint64 {
 	return storage
 }
 
+// StorageCapacity returns retained words for both owned and external storage,
+// without changing the logical bitmap or its ownership.
+func (n *Bitmap) StorageCapacity() int {
+	if n == nil {
+		return 0
+	}
+	return cap(n.data)
+}
+
 func (n *Bitmap) ExternalStorageCapacity() int {
 	if n == nil || !n.HasExternalStorage() {
 		return 0

pkg/common/moerr/error.go

@@ -397,6 +397,7 @@ const (
 	ErrWindowNoInheritFrame                uint16 = 20830
 	ErrWindowNoRedefineOrderBy             uint16 = 20831
 	ErrWindowDuplicateName                 uint16 = 20832
+	ErrScalarSetResourceBudgetExceeded     uint16 = 20833
 
 	// Group 9: streaming
 	ErrUnsupportedOption   uint16 = 20901
@@ -719,6 +720,7 @@ var errorMsgRefer = map[uint16]moErrorMsgItem{
 	ErrTooManyPartitionFuncFields:          {ER_TOO_MANY_PARTITION_FUNC_FIELDS_ERROR, []string{MySQLDefaultSqlState}, "Too many fields in '%-.192s'"},
 	ErrTooManyParameter:                    {ER_PS_MANY_PARAM, []string{MySQLDefaultSqlState}, "Prepared statement contains too many placeholders"},
 	ErrCteMemoryQuotaExceeded:              {ErrCteMemoryQuotaExceeded, []string{MySQLDefaultSqlState}, "recursive CTE memory quota exceeded on this CN: projected %d bytes, query limit %d bytes; increase @@cte_max_memory_bytes or rewrite the query to converge"},
+	ErrScalarSetResourceBudgetExceeded:     {ErrScalarSetResourceBudgetExceeded, []string{MySQLDefaultSqlState}, "scalar set %s budget exceeded: used=%d requested=%d cap=%d"},
 	ErrTooManyWindows:                      {ER_TOO_MANY_WINDOWS, []string{MySQLDefaultSqlState}, "Too many windows in SELECT: %d. Maximum allowed is %d. Use named windows to share windows between window functions."},
 	ErrWindowNoSuchWindow:                  {ER_WINDOW_NO_SUCH_WINDOW, []string{MySQLDefaultSqlState}, "Window name '%s' is not defined."},
 	ErrWindowCircularityInWindowGraph:      {ER_WINDOW_CIRCULARITY_IN_WINDOW_GRAPH, []string{MySQLDefaultSqlState}, "There is a circularity in the window dependency graph."},
@@ -2007,6 +2009,12 @@ func NewCteMemoryQuotaExceeded(ctx context.Context, projected, limit uint64) *Er
 	return newError(ctx, ErrCteMemoryQuotaExceeded, projected, limit)
 }
 
+// Work and admission exhaustion are distinct from OOM: spilling or repeating
+// the same candidate window cannot create more work or concurrency capacity.
+func NewScalarSetResourceBudgetExceeded(ctx context.Context, resource string, used, requested, limit uint64) *Error {
+	return newError(ctx, ErrScalarSetResourceBudgetExceeded, resource, used, requested, limit)
+}
+
 func NewErrTooManyFields(ctx context.Context) *Error {
 	return newError(ctx, ErrTooManyFields)
 }

pkg/common/mpool/allocation_account.go

@@ -138,6 +138,30 @@ type AllocationAccountSnapshot struct {
 	Sealed bool
 }
 
+// RetainedMemoryReservation charges borrowed storage whose physical allocation
+// cannot carry this account. It owns admission only, never the borrowed buffer.
+// Consumers without physical backing provenance may conservatively charge it
+// again; an allocation selection alone is not such provenance.
+type RetainedMemoryReservation struct {
+	account  *AllocationAccount
+	owner    AllocationOwner
+	size     uint64
+	released atomic.Bool
+}
+
+func (a *AllocationAccount) ReserveRetainedMemory(size uint64, owner AllocationOwner) (*RetainedMemoryReservation, error) {
+	if err := a.acquire(size, owner); err != nil {
+		return nil, err
+	}
+	return &RetainedMemoryReservation{account: a, owner: owner, size: size}, nil
+}
+
+func (r *RetainedMemoryReservation) Release() {
+	if r != nil && r.released.CompareAndSwap(false, true) {
+		r.account.release(r.size, r.owner)
+	}
+}
+
 // AllocationAccountOwnerSnapshot is exact, bounded observation state for one
 // repository owner class. Current is live capacity at the snapshot boundary;
 // Peak is that owner's high-water capacity within the generation.

pkg/common/mpool/allocation_account_test.go

@@ -773,3 +773,22 @@ func BenchmarkAllocationAccountCompleteTerminalParallel(b *testing.B) {
 		}
 	})
 }
+
+func TestRetainedMemoryReservationAdmission(t *testing.T) {
+	r, err := NewAllocationAccountRegistry(1, 1)
+	require.NoError(t, err)
+	a, err := r.Open(128)
+	require.NoError(t, err)
+	held, err := a.ReserveRetainedMemory(128, AllocationOwnerWindow)
+	require.NoError(t, err)
+	_, err = a.ReserveRetainedMemory(1, AllocationOwnerWindow)
+	require.ErrorIs(t, err, ErrAllocationAccountCapacity)
+	require.Equal(t, uint64(128), a.Snapshot().Used)
+	held.Release()
+	held.Release()
+	require.Zero(t, a.Snapshot().Used)
+	_, _, err = r.CompleteTerminal(a)
+	require.NoError(t, err)
+	_, err = a.ReserveRetainedMemory(1, AllocationOwnerWindow)
+	require.ErrorIs(t, err, ErrAllocationAccountSealed)
+}

pkg/container/nulls/borrowed_validity.go

@@ -63,6 +63,28 @@ func (nsp *Nulls) BorrowedAccountedBytes() int64 {
 	return nsp.validityLease.AccountedBytes()
 }
 
+// BorrowedRetainedBytes bounds the physical validity backing kept alive by a
+// lease, even when its separate accounting charge is zero.
+func (nsp *Nulls) BorrowedRetainedBytes() uint64 {
+	if nsp == nil || nsp.validityLease == nil {
+		return 0
+	}
+	size := max(uint64(cap(nsp.validity)), uint64(cap(nsp.validityLease.Bytes())))
+	if n := nsp.validityLease.AccountedBytes(); n > 0 {
+		size = max(size, uint64(n))
+	}
+	return size
+}
+
+// RetainedStorageBytes measures owned and borrowed validity storage without
+// crossing the copy-on-write boundary or releasing a lease.
+func (nsp *Nulls) RetainedStorageBytes() uint64 {
+	if nsp == nil {
+		return 0
+	}
+	return uint64(nsp.np.StorageCapacity())*8 + nsp.BorrowedRetainedBytes()
+}
+
 // InitBorrowedWindow creates a retained logical sub-view. It returns false
 // when the requested range contains no NULL, in which case dst remains empty.
 func (nsp *Nulls) InitBorrowedWindow(dst *Nulls, start, end int) (bool, error) {

pkg/container/vector/allocation_account.go

@@ -327,6 +327,42 @@ func (v *Vector) AllocationAccountSelection() *AllocationAccountSelection {
 	return v.allocationAccount
 }
 
+// RetainedBytes conservatively counts storage kept alive by this vector. An
+// allocation selection only governs future allocations and is not backing
+// provenance. Views include the source's complete retained capacity before an
+// offset slice can hide its prefix. Shared storage may be counted twice.
+func (v *Vector) RetainedBytes() uint64 {
+	if v == nil {
+		return 0
+	}
+	metadata := uint64(v.allocatedMetadataBytes())
+	size := retainedBytesAdd(v.retainedPayloadBytes(), metadata)
+	size = retainedBytesAdd(size, v.nsp.RetainedStorageBytes())
+	size = retainedBytesAdd(size, v.gsp.RetainedStorageBytes())
+	return size
+}
+
+func retainedBytesAdd(a, b uint64) uint64 {
+	if b > ^uint64(0)-a {
+		return ^uint64(0)
+	}
+	return a + b
+}
+
+func (v *Vector) retainedPayloadBytes() uint64 {
+	size := uint64(cap(v.data)) + uint64(cap(v.area))
+	for _, lease := range []BufferLease{v.dataLease, v.areaLease} {
+		if lease != nil {
+			n := uint64(cap(lease.Bytes()))
+			if charged := lease.AccountedBytes(); charged > 0 {
+				n = max(n, uint64(charged))
+			}
+			size = retainedBytesAdd(size, n)
+		}
+	}
+	return max(size, v.borrowedBackingBytes)
+}
+
 // CanSetAllocationAccount reports whether selection can be installed without
 // converting or relabeling an existing physical allocation.
 func (v *Vector) CanSetAllocationAccount(

pkg/container/vector/retained_bytes_test.go

@@ -0,0 +1,71 @@
+// 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 vector
+
+import (
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRetainedBytesKeepsOffsetPrefix(t *testing.T) {
+	mp := mpool.MustNewZero()
+	v := NewVec(types.T_int64.ToType())
+	defer v.Free(mp)
+	require.NoError(t, AppendFixedList(v, make([]int64, 1024), nil, mp))
+	original := v.RetainedBytes()
+	w, err := v.Window(1000, 1024)
+	require.NoError(t, err)
+	defer w.Free(mp)
+	for i := 0; i < 10; i++ {
+		next, err := w.Window(1, w.Length())
+		require.NoError(t, err)
+		w.Free(mp)
+		w = next
+	}
+	defer w.Free(mp)
+	require.GreaterOrEqual(t, w.RetainedBytes(), original)
+	require.Equal(t, original, w.RetainedBytes(), "nested views must not accumulate inherited capacity repeatedly")
+	w.Free(mp)
+	require.Zero(t, w.RetainedBytes())
+}
+
+func TestRetainedBytesLeasedSourceLifetime(t *testing.T) {
+	data := make([]byte, 8192)
+	lease, err := NewRefCountedBufferLease(data, 16384, func() {})
+	require.NoError(t, err)
+	defer lease.Release()
+	source, err := NewBorrowedFixedVector(types.T_int64.ToType(), 1, data[8184:], lease)
+	require.NoError(t, err)
+	view, err := source.Window(0, 1)
+	require.NoError(t, err)
+	defer view.Free(nil)
+	source.Free(nil)
+	require.GreaterOrEqual(t, view.RetainedBytes(), uint64(16384))
+	require.Equal(t, int64(0), MustFixedColWithTypeCheck[int64](view)[0])
+}
+
+func TestRetainedBytesOwnedValidityCapacity(t *testing.T) {
+	v := NewVec(types.T_int64.ToType())
+	defer v.Free(nil)
+	v.GetNulls().InitWithSize(65536)
+	v.GetNulls().Add(0)
+	v.SetLength(1)
+	require.GreaterOrEqual(t, v.RetainedBytes(), uint64(8192))
+	v.GetNulls().Reset()
+	require.Zero(t, v.RetainedBytes(), "reset releases ordinary bitmap storage")
+}

pkg/container/vector/vector.go

@@ -121,6 +121,8 @@ type Vector struct {
 	// cantFree bits remain only as the legacy-alias compatibility marker.
 	dataLease BufferLease
 	areaLease BufferLease
+	// Full source capacity retained by legacy views, including offset prefixes.
+	borrowedBackingBytes uint64
 
 	sorted bool // for some optimization
 
@@ -332,29 +334,19 @@ func (v *Vector) Capacity() int {
 // Allocated returns the total allocated memory size of the vector.
 // it can be used to estimate the memory usage of the vector.
 func (v *Vector) Allocated() int {
-	binaryStringBytes := 0
-	if v.binaryStringRows != nil {
-		binaryStringBytes = v.binaryStringRows.Size()
-		if capacity := v.binaryStringRows.ExternalStorageCapacity(); capacity > 0 {
-			binaryStringBytes = 8 * capacity
-		}
-	}
-	if v.textStringRows != nil {
-		textStringBytes := v.textStringRows.Size()
-		if capacity := v.textStringRows.ExternalStorageCapacity(); capacity > 0 {
-			textStringBytes = 8 * capacity
-		}
-		binaryStringBytes += textStringBytes
-	}
-	return cap(v.data) +
-		cap(v.area) +
-		cap(v.prepareParamKinds)*int(unsafe.Sizeof(PrepareParamKind(0))) +
-		cap(v.stringSources)*int(unsafe.Sizeof(types.StringSource(0))) +
-		binaryStringBytes +
+	return cap(v.data) + cap(v.area) + v.allocatedMetadataBytes() +
 		8*v.nsp.GetBitmap().ExternalStorageCapacity() +
 		8*v.gsp.GetBitmap().ExternalStorageCapacity()
 }
 
+// allocatedMetadataBytes excludes payload and validity storage. Unlike legacy
+// bitmap access, this measurement does not materialize borrowed validity.
+func (v *Vector) allocatedMetadataBytes() int {
+	return cap(v.prepareParamKinds)*int(unsafe.Sizeof(PrepareParamKind(0))) +
+		cap(v.stringSources)*int(unsafe.Sizeof(types.StringSource(0))) +
+		8*v.binaryStringRows.StorageCapacity() + 8*v.textStringRows.StorageCapacity()
+}
+
 func (v *Vector) SetLength(n int) {
 	if n < 0 {
 		panic("negative vector length")
@@ -4435,6 +4427,7 @@ func (v *Vector) Free(mp *mpool.MPool) {
 	v.cantFreeArea = false
 	v.dataLease = nil
 	v.areaLease = nil
+	v.borrowedBackingBytes = 0
 
 	v.nsp.Reset()
 	v.gsp.Reset()
@@ -4986,6 +4979,7 @@ func (v *Vector) unmarshalBinary(data []byte, validateValues bool) error {
 	v.length = layout.length
 	v.data = layout.data
 	v.area = layout.area
+	v.borrowedBackingBytes = uint64(cap(data))
 	v.nsp = decodedNulls
 	v.gsp.Reset()
 	v.sorted = layout.sorted
@@ -10536,6 +10530,7 @@ func (v *Vector) window(
 	w.class = v.class
 	w.length = end - start
 	w.sorted = v.sorted
+	w.borrowedBackingBytes = v.retainedPayloadBytes()
 	if err := v.copyBinaryStringWindowTo(w, start, end, mp); err != nil {
 		w.Free(mp)
 		return nil, err

pkg/defines/const.go

@@ -162,7 +162,8 @@ const (
 	MORPCVersion124    int64 = 124 // exact fractional TIMESTAMPADD SECOND intervals
 	MORPCVersion125    int64 = 125 // LOAD IGNORE foreign-key row rejection diagnostics
 	MORPCVersion126    int64 = 126 // parser-derived information_schema.VIEWS definitions
-	MORPCLatestVersion       = MORPCVersion126
+	MORPCVersion127    int64 = 127 // signed-zero set keys and bounded local scalar-set reaggregation
+	MORPCLatestVersion       = MORPCVersion127
 )
 
 // DefaultLockWaitTimeoutSeconds is shared by the frontend default and by

pkg/pb/pipeline/iceberg_redact_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/gogo/protobuf/protoc-gen-gogo/descriptor"
 	"github.com/matrixorigin/matrixone/pkg/iceberg/testutil"
+	planpb "github.com/matrixorigin/matrixone/pkg/pb/plan"
 )
 
 func TestIcebergRedactionDisablesGeneratedStringers(t *testing.T) {
@@ -423,3 +424,62 @@ func assertRedactedString(t *testing.T, got string, forbidden ...string) {
 func containsRedactionMarker(got string) bool {
 	return testutil.HasIcebergRedactionMarker(got)
 }
+
+func TestPipelineDescriptorMergedScalarAndLockContracts(t *testing.T) {
+	want := map[string]map[string]int32{
+		"Instruction":    {"scalar_set_reaggregation": 65},
+		"DedupJoin":      {"dedup_key_exprs": 30},
+		"RightDedupJoin": {"dedup_key_exprs": 18},
+		"HashBuild":      {"dedup_key_exprs": 20},
+		"LockTarget":     {"no_wait": 13, "skip_locked": 14},
+	}
+	for _, message := range decodePipelineFileDescriptor(t).GetMessageType() {
+		fields, ok := want[message.GetName()]
+		if !ok {
+			continue
+		}
+		for _, field := range message.GetField() {
+			if number, exists := fields[field.GetName()]; exists {
+				if field.GetNumber() != number {
+					t.Fatalf("%s.%s = %d, want %d", message.GetName(), field.GetName(), field.GetNumber(), number)
+				}
+				delete(fields, field.GetName())
+			}
+		}
+		if len(fields) != 0 {
+			t.Fatalf("missing %s fields: %v", message.GetName(), fields)
+		}
+		delete(want, message.GetName())
+	}
+	if len(want) != 0 {
+		t.Fatalf("missing merged messages: %v", want)
+	}
+}
+
+func TestMergedDiagnosticPipelineExpressionAdmission(t *testing.T) {
+	expr := &planpb.Expr{Typ: planpb.Type{Id: 28}, Expr: &planpb.Expr_F{F: &planpb.Function{Func: &planpb.ObjectRef{Obj: 605 << 32}}}}
+	for _, instruction := range []*Instruction{
+		{ScalarSetReaggregation: true, DedupJoin: &DedupJoin{DedupKeyExprs: []*planpb.Expr{expr}}},
+		{ScalarSetReaggregation: true, RightDedupJoin: &RightDedupJoin{DedupKeyExprs: []*planpb.Expr{expr}}},
+		{ScalarSetReaggregation: true, HashBuild: &HashBuild{DedupKeyExprs: []*planpb.Expr{expr}}},
+	} {
+		data, err := instruction.Marshal()
+		if err != nil {
+			t.Fatal(err)
+		}
+		var decoded Instruction
+		if err := decoded.Unmarshal(data); err != nil {
+			t.Fatal(err)
+		}
+		if !proto.Equal(instruction, &decoded) {
+			t.Fatal("diagnostic expressions and scalar marker must coexist on wire")
+		}
+		features, err := planpb.RequiredRemoteExpressionFeatures(&decoded)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !features.SetOperationFloatKeys || !features.ScalarSetReaggregation {
+			t.Fatalf("missing merged expression admission: %+v", features)
+		}
+	}
+}