Repository: matrixorigin/matrixone
Pull request #12859 (2023-11-21 07:39:55)

improve mem usage of PrimaryKeysMayBeModified
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue https://github.com/matrixorigin/MO-Cloud/issues/1574

## What this PR does / why we need it:
improve mem usage of PrimaryKeysMayBeModified
在处理 https://github.com/matrixorigin/matrixone/pull/12818  这个的时候。
在profile中看到这个方法占用时间会多一些。所以做了简单的优化。
主要是改了一下函数参数，把一个batch的数据都传进去，这样部分函数就只需要执行一次了。主要是这个
p.primaryIndex.Copy().Iter()   主要是这个一个批次只要执行一次了。
另外就是去掉了  p.rows.Copy().Iter() 的Copy()


Patch:
diff --git a/pkg/vm/engine/disttae/logtailreplay/primary_key_utils.go b/pkg/vm/engine/disttae/logtailreplay/primary_key_utils.go
index 1ba082a83..71025c8be 100644
--- a/pkg/vm/engine/disttae/logtailreplay/primary_key_utils.go
+++ b/pkg/vm/engine/disttae/logtailreplay/primary_key_utils.go
@@ -24,7 +24,7 @@ import (
 func (p *PartitionState) PrimaryKeyMayBeModified(
 	from types.TS,
 	to types.TS,
-	key []byte,
+	keys [][]byte,
 ) bool {
 
 	p.shared.Lock()
@@ -38,56 +38,60 @@ func (p *PartitionState) PrimaryKeyMayBeModified(
 	}
 
 	iter := p.primaryIndex.Copy().Iter()
+	pivot := RowEntry{
+		Time: types.BuildTS(math.MaxInt64, math.MaxUint32),
+	}
+	idxEntry := &PrimaryIndexEntry{}
 	defer iter.Release()
 
-	for ok := iter.Seek(&PrimaryIndexEntry{
-		Bytes: key,
-	}); ok; ok = iter.Next() {
+	for _, key := range keys {
 
-		entry := iter.Item()
+		idxEntry.Bytes = key
 
-		if !bytes.Equal(entry.Bytes, key) {
-			break
-		}
+		for ok := iter.Seek(idxEntry); ok; ok = iter.Next() {
 
-		if entry.Time.GreaterEq(from) {
-			return true
-		}
+			entry := iter.Item()
 
-		// some legacy deletion entries may not indexed, check all rows for changes
-		pivot := RowEntry{
-			BlockID: entry.BlockID,
-			RowID:   entry.RowID,
-			Time:    types.BuildTS(math.MaxInt64, math.MaxUint32),
-		}
-		iter := p.rows.Copy().Iter()
-		seek := false
-		for {
-			if !seek {
-				seek = true
-				if !iter.Seek(pivot) {
+			if !bytes.Equal(entry.Bytes, key) {
+				break
+			}
+
+			if entry.Time.GreaterEq(from) {
+				return true
+			}
+
+			// some legacy deletion entries may not indexed, check all rows for changes
+			pivot.BlockID = entry.BlockID
+			pivot.RowID = entry.RowID
+			rowIter := p.rows.Iter()
+			seek := false
+			for {
+				if !seek {
+					seek = true
+					if !rowIter.Seek(pivot) {
+						break
+					}
+				} else {
+					if !rowIter.Next() {
+						break
+					}
+				}
+				row := rowIter.Item()
+				if row.BlockID.Compare(entry.BlockID) != 0 {
 					break
 				}
-			} else {
-				if !iter.Next() {
+				if !row.RowID.Equal(entry.RowID) {
 					break
 				}
+				if row.Time.GreaterEq(from) {
+					rowIter.Release()
+					return true
+				}
 			}
-			row := iter.Item()
-			if row.BlockID.Compare(entry.BlockID) != 0 {
-				break
-			}
-			if !row.RowID.Equal(entry.RowID) {
-				break
-			}
-			if row.Time.GreaterEq(from) {
-				iter.Release()
-				return true
-			}
+			rowIter.Release()
 		}
-		iter.Release()
 
+		iter.First()
 	}
-
 	return false
 }
diff --git a/pkg/vm/engine/disttae/txn_table.go b/pkg/vm/engine/disttae/txn_table.go
index 1fb78aed6..92eeac260 100644
--- a/pkg/vm/engine/disttae/txn_table.go
+++ b/pkg/vm/engine/disttae/txn_table.go
@@ -1885,12 +1885,9 @@ func (tbl *txnTable) PrimaryKeysMayBeModified(ctx context.Context, from types.TS
 	packer.Reset()
 
 	keys := logtailreplay.EncodePrimaryKeyVector(keysVector, packer)
-	for _, key := range keys {
-		if snap.PrimaryKeyMayBeModified(from, to, key) {
-			return true, nil
-		}
+	if snap.PrimaryKeyMayBeModified(from, to, keys) {
+		return true, nil
 	}
-
 	return false, nil
 }
 


Test patch:
diff --git a/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go b/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go
index f1a015434..edc626e15 100644
--- a/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go
+++ b/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go
@@ -112,7 +112,7 @@ func TestPartitionStateRowsIter(t *testing.T) {
 		}
 		require.Equal(t, 1, n)
 		require.Nil(t, iter.Close())
-		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), bs)
+		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), [][]byte{bs})
 		require.True(t, modified)
 	}
 
@@ -207,7 +207,7 @@ func TestPartitionStateRowsIter(t *testing.T) {
 			modified := state.PrimaryKeyMayBeModified(
 				types.BuildTS(int64(deleteAt+i), 0).Prev(),
 				types.BuildTS(int64(deleteAt+i), 0).Next(),
-				key,
+				[][]byte{key},
 			)
 			require.True(t, modified)
 		}
@@ -349,7 +349,7 @@ func TestInsertAndDeleteAtTheSameTimestamp(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ts := types.BuildTS(int64(i), 0)
 		key := EncodePrimaryKey(int64(i), packer)
-		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), key)
+		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), [][]byte{key})
 		require.True(t, modified)
 	}
 
@@ -441,7 +441,7 @@ func TestDeleteBeforeInsertAtTheSameTime(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ts := types.BuildTS(int64(i), 0)
 		key := EncodePrimaryKey(int64(i), packer)
-		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), key)
+		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), [][]byte{key})
 		require.True(t, modified)
 	}
 
@@ -486,7 +486,7 @@ func TestPrimaryKeyModifiedWithDeleteOnly(t *testing.T) {
 	for i := 0; i < num; i++ {
 		ts := types.BuildTS(int64(i), 0)
 		key := EncodePrimaryKey(int64(i), packer)
-		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), key)
+		modified := state.PrimaryKeyMayBeModified(ts.Prev(), ts.Next(), [][]byte{key})
 		require.True(t, modified)
 	}