fix changes handle (#23250)

fix changes handle

Approved by: @XuPeng-SH

Author: jiangxinmeng1

pkg/vm/engine/disttae/logtailreplay/change_handle.go

@@ -1471,7 +1471,7 @@ func updateDataBatch(bat *batch.Batch, start, end types.TS, mp *mpool.MPool) {
 func updateCNTombstoneBatch(bat *batch.Batch, committs types.TS, mp *mpool.MPool) {
 	var pk *vector.Vector
 	for _, vec := range bat.Vecs {
-		if vec.GetType().Oid != types.T_Rowid {
+		if vec.GetType().Oid != types.T_Rowid || vec.GetType().Oid != types.T_TS {
 			pk = vec
 		} else {
 			vec.Free(mp)
@@ -1485,6 +1485,12 @@ func updateCNTombstoneBatch(bat *batch.Batch, committs types.TS, mp *mpool.MPool
 	bat.Attrs = []string{objectio.TombstoneAttr_PK_Attr, objectio.DefaultCommitTS_Attr}
 }
 func updateCNDataBatch(bat *batch.Batch, commitTS types.TS, mp *mpool.MPool) {
+	for i, vec := range bat.Vecs {
+		if vec.GetType().Oid == types.T_TS {
+			vec.Free(mp)
+			bat.Vecs = append(bat.Vecs[:i], bat.Vecs[i+1:]...)
+		}
+	}
 	commitTSVec, err := vector.NewConstFixed(types.T_TS.ToType(), commitTS, bat.Vecs[0].Length(), mp)
 	if err != nil {
 		return

pkg/vm/engine/disttae/logtailreplay/change_handle_test.go

@@ -0,0 +1,110 @@
+// Copyright 2024 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 logtailreplay
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+)
+
+func TestUpdateCNDataBatch_RemoveTSVector(t *testing.T) {
+	mp := mpool.MustNewZero()
+
+	// Create a batch with multiple vectors including T_TS type
+	bat := batch.NewWithSize(3)
+
+	// Create an int64 vector (non-TS)
+	vec1 := vector.NewVec(types.T_int64.ToType())
+	vector.AppendFixed(vec1, int64(1), false, mp)
+	vector.AppendFixed(vec1, int64(2), false, mp)
+	bat.Vecs[0] = vec1
+
+	// Create a T_TS vector (should be removed)
+	tsVec1 := vector.NewVec(types.T_TS.ToType())
+	vector.AppendFixed(tsVec1, types.BuildTS(1, 0), false, mp)
+	vector.AppendFixed(tsVec1, types.BuildTS(2, 0), false, mp)
+	bat.Vecs[1] = tsVec1
+
+	// Create another int64 vector (non-TS)
+	vec2 := vector.NewVec(types.T_int64.ToType())
+	vector.AppendFixed(vec2, int64(3), false, mp)
+	vector.AppendFixed(vec2, int64(4), false, mp)
+	bat.Vecs[2] = vec2
+
+	bat.SetRowCount(2)
+
+	// Verify initial state: should have 3 vectors
+	require.Equal(t, 3, len(bat.Vecs))
+	require.Equal(t, types.T_int64, bat.Vecs[0].GetType().Oid)
+	require.Equal(t, types.T_TS, bat.Vecs[1].GetType().Oid)
+	require.Equal(t, types.T_int64, bat.Vecs[2].GetType().Oid)
+
+	// Call updateCNDataBatch
+	newCommitTS := types.BuildTS(100, 0)
+	updateCNDataBatch(bat, newCommitTS, mp)
+
+	// Verify T_TS vector is removed and new commitTS vector is added at the end
+	require.Equal(t, 3, len(bat.Vecs))
+	require.Equal(t, types.T_int64, bat.Vecs[0].GetType().Oid)
+	require.Equal(t, types.T_int64, bat.Vecs[1].GetType().Oid)
+	require.Equal(t, types.T_TS, bat.Vecs[2].GetType().Oid)
+
+	// Verify the new commitTS vector has the correct value
+	require.True(t, bat.Vecs[2].IsConst())
+	tsVal := vector.MustFixedColWithTypeCheck[types.TS](bat.Vecs[2])[0]
+	require.Equal(t, newCommitTS, tsVal)
+
+	bat.Clean(mp)
+}
+
+func TestUpdateCNDataBatch_NoTSVector(t *testing.T) {
+	mp := mpool.MustNewZero()
+
+	// Create a batch without T_TS vectors
+	bat := batch.NewWithSize(2)
+
+	vec1 := vector.NewVec(types.T_int64.ToType())
+	vector.AppendFixed(vec1, int64(1), false, mp)
+	vector.AppendFixed(vec1, int64(2), false, mp)
+	bat.Vecs[0] = vec1
+
+	vec2 := vector.NewVec(types.T_int64.ToType())
+	vector.AppendFixed(vec2, int64(3), false, mp)
+	vector.AppendFixed(vec2, int64(4), false, mp)
+	bat.Vecs[1] = vec2
+
+	bat.SetRowCount(2)
+
+	// Verify initial state: should have 2 vectors
+	require.Equal(t, 2, len(bat.Vecs))
+
+	// Call updateCNDataBatch
+	newCommitTS := types.BuildTS(100, 0)
+	updateCNDataBatch(bat, newCommitTS, mp)
+
+	// Verify commitTS vector is added at the end
+	require.Equal(t, 3, len(bat.Vecs))
+	require.Equal(t, types.T_int64, bat.Vecs[0].GetType().Oid)
+	require.Equal(t, types.T_int64, bat.Vecs[1].GetType().Oid)
+	require.Equal(t, types.T_TS, bat.Vecs[2].GetType().Oid)
+
+	bat.Clean(mp)
+}

pkg/vm/engine/tae/txn/txnbase/mvccnode.go

@@ -21,7 +21,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
-	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/txnif"
 )
@@ -79,7 +78,6 @@ func (un *TxnMVCCNode) CheckConflict(txn txnif.TxnReader) error {
 		if un.IsSameTxn(txn) {
 			return nil
 		}
-		logutil.Infof("Dedup-WW, prev txn is %x, new txn is %x", un.Txn.GetID(), txn.GetID())
 		return txnif.ErrTxnWWConflict
 	}
 
@@ -88,7 +86,6 @@ func (un *TxnMVCCNode) CheckConflict(txn txnif.TxnReader) error {
 	//        ts         CommitTs            time
 	startTS := txn.GetStartTS()
 	if un.End.GT(&startTS) {
-		logutil.Infof("Dedup-WW, prev ts is %x, new txn is %x", un.End.ToString(), startTS.ToString())
 		return txnif.ErrTxnWWConflict
 	}
 	return nil