clean some code in the workspace. (#21725)

Clean some code in the workspace.
1. gen new row id
2. deletion type.

Approved by: @aunjgr, @XuPeng-SH, @ouyuanning

Author: gouhongshen

pkg/container/batch/batch.go

@@ -362,6 +362,19 @@ func (bat *Batch) SetAttributes(attrs []string) {
 	bat.Attrs = attrs
 }
 
+func (bat *Batch) InsertVector(
+	pos int32,
+	attr string,
+	vec *vector.Vector,
+) {
+	bat.Vecs = append(bat.Vecs, nil)
+	copy(bat.Vecs[pos+1:], bat.Vecs[pos:])
+	bat.Vecs[pos] = vec
+	bat.Attrs = append(bat.Attrs, "")
+	copy(bat.Attrs[pos+1:], bat.Attrs[pos:])
+	bat.Attrs[pos] = attr
+}
+
 func (bat *Batch) SetVector(pos int32, vec *vector.Vector) {
 	bat.Vecs[pos] = vec
 	if vec != nil {

pkg/container/types/rowid.go

@@ -18,9 +18,11 @@ import (
 	"bytes"
 	"encoding/hex"
 	"fmt"
+	"math"
 	"unsafe"
 
 	"github.com/google/uuid"
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/util"
 )
 
@@ -198,6 +200,18 @@ func (r *Rowid) SetRowOffset(offset uint32) {
 	copy(r[BlockidSize:], EncodeUint32(&offset))
 }
 
+func (r *Rowid) SetBlkOffset(offset uint16) {
+	copy(r[ObjectidSize:], EncodeUint16(&offset))
+}
+
+func (r *Rowid) SetObjOffset(offset uint16) {
+	copy(r[SegmentidSize:], EncodeUint16(&offset))
+}
+
+func (r *Rowid) SetSegment(seg Segmentid) {
+	copy(r[:SegmentidSize], seg[:])
+}
+
 func (r *Rowid) GetRowOffset() uint32 {
 	return DecodeUint32(r[BlockidSize:])
 }
@@ -206,6 +220,10 @@ func (r *Rowid) GetBlockOffset() uint16 {
 	return DecodeUint16(r[ObjectBytesSize:BlockidSize])
 }
 
+func (r *Rowid) GetObjectOffset() uint16 {
+	return DecodeUint16(r[SegmentidSize:ObjectBytesSize])
+}
+
 func (r *Rowid) GetObjectString() string {
 	uuid := (*uuid.UUID)(r[:UuidSize])
 	s := DecodeUint16(r[UuidSize:ObjectBytesSize])
@@ -224,6 +242,38 @@ func (r *Rowid) ShortStringEx() string {
 	return fmt.Sprintf("%s-%d", b.ShortStringEx(), s)
 }
 
+func (r *Rowid) IncrBlk() error {
+	blkOffset := r.GetBlockOffset()
+	if blkOffset == math.MaxUint16 {
+		if err := r.IncrObj(); err != nil {
+			return err
+		}
+		blkOffset = 0
+	} else {
+		blkOffset++
+	}
+
+	r.SetBlkOffset(blkOffset)
+	r.SetRowOffset(0)
+
+	return nil
+}
+
+func (r *Rowid) IncrObj() error {
+	objOffset := r.GetObjectOffset()
+	if objOffset == math.MaxUint16 {
+		// we expect that the segment id of a rowId is immutable, so incr the segment
+		// cannot fix the object overflow.
+		return moerr.NewInternalErrorNoCtxf("rowId object offset overflow, curr rowId: %s", r.String())
+	}
+
+	objOffset++
+	r.SetObjOffset(objOffset)
+	r.SetBlkOffset(0)
+	r.SetRowOffset(0)
+	return nil
+}
+
 func (b *Blockid) EQ(than *Blockid) bool {
 	return b.Compare(than) == 0
 }

pkg/container/types/types.go

@@ -15,6 +15,7 @@
 package types
 
 import (
+	"bytes"
 	"encoding/binary"
 	"fmt"
 
@@ -257,6 +258,10 @@ type Rowid [RowidSize]byte
 // Segmentid
 type Segmentid = Uuid
 
+func (d *Segmentid) EQ(oth *Segmentid) bool {
+	return bytes.Equal(d[:], oth[:])
+}
+
 // Objectid
 type Objectid [ObjectidSize]byte
 

pkg/sql/colexec/cn_segment_map.go

@@ -0,0 +1,95 @@
+// Copyright 2025 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 colexec
+
+import (
+	"sync"
+
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/objectio"
+)
+
+const (
+	TxnWorkspaceUnCommitType = 1
+)
+
+var (
+	// TxnWorkspaceSegment includes a dedicated header which indicates it is a workspace segment.
+	TxnWorkspaceSegment = types.Segmentid([16]byte{
+		0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0xFF,
+	})
+)
+
+type CnSegmentMap struct {
+	sync.Mutex
+	// tag whether a segment is generated by this txn
+	mp map[objectio.Segmentid]int32
+}
+
+func IsDeletionOnTxnUnCommit(
+	segmentMap map[string]int32,
+	segId *types.Segmentid,
+) bool {
+	// raw row id generated by txn.
+	// WriteBatch()
+	if segId.EQ(&TxnWorkspaceSegment) {
+		return true
+	}
+
+	// check if it is a persisted segment generated by a workspace.
+	if segmentMap != nil {
+		return segmentMap[string(segId[:])] == TxnWorkspaceUnCommitType
+	} else {
+		ss := Get()
+		if ss == nil {
+			return false
+		}
+
+		return ss.GetCnSegmentType(segId) == TxnWorkspaceUnCommitType
+	}
+}
+
+// IsDeletionOnTxnUnCommitPersisted check if the segId
+// is an UnCommit and flushed segment in the workspace.
+// if the input map is nil, this check will be done by the colexec.Get.
+func IsDeletionOnTxnUnCommitPersisted(
+	segmentMap map[string]int32,
+	segId *types.Segmentid,
+) bool {
+	if IsDeletionOnTxnUnCommit(segmentMap, segId) {
+		return !segId.EQ(&TxnWorkspaceSegment)
+	}
+	return false
+}
+
+// IsDeletionOnTxnUnCommitInMem check if the segId
+// is an UnCommit inmem segment in the workspace.
+// if the input map is nil, this check will be done by the colexec.Get.
+func IsDeletionOnTxnUnCommitInMem(
+	segmentMap map[string]int32,
+	segId *types.Segmentid,
+) bool {
+	if IsDeletionOnTxnUnCommit(segmentMap, segId) {
+		return segId.EQ(&TxnWorkspaceSegment)
+	}
+	return false
+}
+
+func RecordTxnUnCommitSegment(segId *types.Segmentid) {
+	Get().PutCnSegment(segId, TxnWorkspaceUnCommitType)
+}

pkg/sql/colexec/deletion/deletion.go

@@ -31,21 +31,11 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
-//row id be divided into four types:
-// 1. RawBatchOffset : belong to txn's workspace
-// 2. CNBlockOffset  : belong to txn's workspace
-
-// 3. RawRowIdBatch  : belong to txn's snapshot data.
-// 4. FlushDeltaLoc   : belong to txn's snapshot data, which on S3 and pointed by delta location.
 const (
-	RawRowIdBatch = iota
-	// remember that, for one block,
-	// when it sends the info to mergedeletes,
-	// either it's Compaction or not.
-	Compaction
-	CNBlockOffset
-	RawBatchOffset
-	FlushDeltaLoc
+	FlushDeltaLoc = iota
+
+	DeletionOnTxnUnCommit
+	DeletionOnCommitted
 )
 
 const opName = "deletion"

pkg/sql/colexec/deletion/types.go

@@ -236,7 +236,7 @@ func (ctr *container) flush(proc *process.Process, analyzer process.Analyzer) (u
 		blkids := make([]types.Blockid, 0, len(blockId_rowIdBatch))
 		for blkid := range blockId_rowIdBatch {
 			//Don't flush rowids belong to uncommitted cn block and raw data batch in txn's workspace.
-			if ctr.blockId_type[blkid] != RawRowIdBatch {
+			if ctr.blockId_type[blkid] == DeletionOnTxnUnCommit {
 				continue
 			}
 			blkids = append(blkids, blkid)
@@ -329,12 +329,10 @@ func collectBatchInfo(proc *process.Process, deletion *Deletion, destBatch *batc
 
 		deletion.ctr.deleted_length += 1
 
-		if deletion.SegmentMap[string(segid[:])] == colexec.TxnWorkSpaceIdType {
-			deletion.ctr.blockId_type[blkid] = RawBatchOffset
-		} else if deletion.SegmentMap[string(segid[:])] == colexec.CnBlockIdType {
-			deletion.ctr.blockId_type[blkid] = CNBlockOffset
+		if colexec.IsDeletionOnTxnUnCommit(deletion.SegmentMap, &segid) {
+			deletion.ctr.blockId_type[blkid] = DeletionOnTxnUnCommit
 		} else {
-			deletion.ctr.blockId_type[blkid] = RawRowIdBatch
+			deletion.ctr.blockId_type[blkid] = DeletionOnCommitted
 		}
 
 		if _, ok := deletion.ctr.partitionId_blockId_rowIdBatch[pIdx]; !ok {

pkg/sql/colexec/mergedelete/mergedelete_test.go

@@ -27,7 +27,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
-	"github.com/matrixorigin/matrixone/pkg/sql/colexec/deletion"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 	"github.com/matrixorigin/matrixone/pkg/vm"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
@@ -81,7 +80,7 @@ func TestMergeDelete(t *testing.T) {
 		Vecs: []*vector.Vector{
 			testutil.MakeTextVector([]string{"mock_block_id0"}, nil),
 			testutil.MakeTextVector([]string{string(bytes)}, nil),
-			testutil.MakeInt8Vector([]int8{deletion.RawBatchOffset}, nil),
+			testutil.MakeInt8Vector([]int8{0}, nil),
 			testutil.MakeInt32Vector([]int32{0}, nil),
 			vcu32,
 		},
@@ -150,7 +149,7 @@ func TestMergeDelete(t *testing.T) {
 		Vecs: []*vector.Vector{
 			testutil.MakeTextVector([]string{"mock_block_id1", "mock_block_id2", "mock_block_id3"}, nil),
 			testutil.MakeTextVector([]string{string(bytes1), string(bytes2), string(bytes3)}, nil),
-			testutil.MakeInt8Vector([]int8{deletion.RawRowIdBatch, deletion.CNBlockOffset, deletion.FlushDeltaLoc}, nil),
+			testutil.MakeInt8Vector([]int8{0, 1, 2}, nil),
 			testutil.MakeInt32Vector([]int32{0, 0, 0}, nil),
 			vcu32_2,
 		},

pkg/sql/colexec/multi_update/s3writer_delegate.go

@@ -70,7 +70,7 @@ type deleteBlockInfo struct {
 func newDeleteBlockData(inputBatch *batch.Batch, pkIdx int) *deleteBlockData {
 	data := &deleteBlockData{
 		bitmap: nulls.NewWithSize(int(options.DefaultBlockMaxRows)),
-		typ:    deletion.RawRowIdBatch,
+		typ:    deletion.DeletionOnCommitted,
 		bat:    newDeleteBatch(inputBatch, pkIdx),
 	}
 	return data
@@ -218,13 +218,11 @@ func (writer *s3WriterDelegate) prepareDeleteBatchs(
 				if err != nil {
 					return nil, err
 				}
-				strSegid := string(segid[:])
-				if writer.segmentMap[strSegid] == colexec.TxnWorkSpaceIdType {
-					blockMap[blkid].typ = deletion.RawBatchOffset
-				} else if writer.segmentMap[strSegid] == colexec.CnBlockIdType {
-					blockMap[blkid].typ = deletion.CNBlockOffset
+
+				if colexec.IsDeletionOnTxnUnCommit(writer.segmentMap, &segid) {
+					blockMap[blkid].typ = deletion.DeletionOnTxnUnCommit
 				} else {
-					blockMap[blkid].typ = deletion.RawRowIdBatch
+					blockMap[blkid].typ = deletion.DeletionOnCommitted
 				}
 			}
 
@@ -247,7 +245,7 @@ func (writer *s3WriterDelegate) prepareDeleteBatchs(
 	blkids := make([]types.Blockid, 0, len(blockMap))
 	for blkid, data := range blockMap {
 		//Don't flush rowids belong to uncommitted cn block and raw data batch in txn's workspace.
-		if data.typ != deletion.RawRowIdBatch {
+		if data.typ == deletion.DeletionOnTxnUnCommit {
 			continue
 		}
 		blkids = append(blkids, blkid)

pkg/sql/colexec/server.go

@@ -27,11 +27,6 @@ import (
 // FIXME: shit design
 var srv atomic.Pointer[Server]
 
-const (
-	TxnWorkSpaceIdType = 1
-	CnBlockIdType      = 2
-)
-
 func Get() *Server {
 	return srv.Load()
 }

pkg/sql/colexec/types.go

@@ -26,7 +26,6 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/matrixorigin/matrixone/pkg/logservice"
-	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 )
 
@@ -107,15 +106,6 @@ type UuidProcMap struct {
 	mp map[uuid.UUID]uuidProcMapItem
 }
 
-type CnSegmentMap struct {
-	sync.Mutex
-	// tag whether a segment is generated by this txn
-	// segmentName => uuid + file number
-	// 1.mp[segmentName] = 1 => txnWorkSpace
-	// 2.mp[segmentName] = 2 => Cn Blcok
-	mp map[objectio.Segmentid]int32
-}
-
 const (
 	DefaultBatchSize = 8192
 )