fix data copy in replaying from checkpoint - main (#21233)

- Copy bytes to target fields

Approved by: @XuPeng-SH

Author: aptend

pkg/container/vector/vector.go

@@ -278,6 +278,13 @@ func GetFixedAtWithTypeCheck[T any](v *Vector, idx int) T {
 	return slice[idx]
 }
 
+func (v *Vector) CloneBytesAt(i int) []byte {
+	bs := v.GetBytesAt(i)
+	ret := make([]byte, len(bs))
+	copy(ret, bs)
+	return ret
+}
+
 func (v *Vector) GetBytesAt(i int) []byte {
 	if v.IsConst() {
 		i = 0

pkg/vm/engine/tae/catalog/catalogreplay.go

@@ -387,12 +387,13 @@ func (catalog *Catalog) ReplayMOTables(ctx context.Context, txnNode *txnbase.Txn
 			schema.Relkind = tblBat.GetVectorByName(pkgcatalog.SystemRelAttr_Kind).GetDownstreamVector().GetStringAt(i)
 			schema.Createsql = tblBat.GetVectorByName(pkgcatalog.SystemRelAttr_CreateSQL).GetDownstreamVector().GetStringAt(i)
 			schema.View = tblBat.GetVectorByName(pkgcatalog.SystemRelAttr_ViewDef).GetDownstreamVector().GetStringAt(i)
-			schema.Constraint = tblBat.GetVectorByName(pkgcatalog.SystemRelAttr_Constraint).GetDownstreamVector().GetBytesAt(i)
+			schema.Constraint = tblBat.GetVectorByName(pkgcatalog.SystemRelAttr_Constraint).GetDownstreamVector().CloneBytesAt(i)
 			schema.AcInfo = accessInfo{}
 			schema.AcInfo.RoleID = roleIDs[i]
 			schema.AcInfo.UserID = userIDs[i]
 			schema.AcInfo.CreateAt = createAts[i]
 			schema.AcInfo.TenantID = tenantIDs[i]
+			// unmarshal before releasing, no need to copy
 			extra := tblBat.GetVectorByName(pkgcatalog.SystemRelAttr_ExtraInfo).GetDownstreamVector().GetBytesAt(i)
 			schema.MustRestoreExtra(extra)
 			if err := schema.Finalize(true); err != nil {

pkg/vm/engine/tae/catalog/schema.go

@@ -622,6 +622,7 @@ func (s *Schema) ReadFromBatch(
 		}
 		def := new(ColDef)
 		def.Name = bat.GetVectorByName((pkgcatalog.SystemColAttr_Name)).GetDownstreamVector().GetStringAt(offset)
+		// unmarshal before releasing, no need to copy
 		data := bat.GetVectorByName((pkgcatalog.SystemColAttr_Type)).GetDownstreamVector().GetBytesAt(offset)
 		types.Decode(data, &def.Type)
 		nullable := nullables[offset]
@@ -636,8 +637,8 @@ func (s *Schema) ReadFromBatch(
 		isAutoIncrement := autoIncrements[offset]
 		def.AutoIncrement = i82bool(isAutoIncrement)
 		def.Comment = bat.GetVectorByName((pkgcatalog.SystemColAttr_Comment)).GetDownstreamVector().GetStringAt(offset)
-		def.OnUpdate = bat.GetVectorByName((pkgcatalog.SystemColAttr_Update)).GetDownstreamVector().GetBytesAt(offset)
-		def.Default = bat.GetVectorByName((pkgcatalog.SystemColAttr_DefaultExpr)).GetDownstreamVector().GetBytesAt(offset)
+		def.OnUpdate = bat.GetVectorByName((pkgcatalog.SystemColAttr_Update)).GetDownstreamVector().CloneBytesAt(offset)
+		def.Default = bat.GetVectorByName((pkgcatalog.SystemColAttr_DefaultExpr)).GetDownstreamVector().CloneBytesAt(offset)
 		def.Idx = int(idxes[offset]) - 1
 		def.SeqNum = seqNums[offset]
 		def.EnumValues = bat.GetVectorByName((pkgcatalog.SystemColAttr_EnumValues)).GetDownstreamVector().GetStringAt(offset)