adjust debug log (#19696)

- lower log print threshold
- check rowid of insert batch

Approved by: @LeftHandCold, @XuPeng-SH, @sukki37

Author: aptend

pkg/objectio/injects.go

@@ -30,13 +30,19 @@ const (
 	FJ_TracePartitionState = "fj/trace/partitionstate"
 
 	FJ_Debug19524 = "fj/debug/19524"
+	FJ_Debug19357 = "fj/debug/19357"
 )
 
 func Debug19524Injected() bool {
 	_, _, injected := fault.TriggerFault(FJ_Debug19524)
 	return injected
 }
 
+func Debug19357Injected() (bool, int64) {
+	iret, _, injected := fault.TriggerFault(FJ_Debug19357)
+	return injected, iret
+}
+
 func RangesInjected(name string) bool {
 	_, sarg, injected := fault.TriggerFault(FJ_TraceRanges)
 	if !injected {

pkg/vm/engine/disttae/local_disttae_datasource.go

@@ -406,39 +406,34 @@ func checkWorkspaceEntryType(
 	return (entry.typ == DELETE) && (entry.fileName == "")
 }
 
-func checkTxnOffsetZero(ls *LocalDisttaeDataSource, writes []Entry) {
-	if len(writes) > 200 && ls.txnOffset == 0 && ls.table.accountId == 0 && ls.table.tableName == "mo_increment_columns" {
-		logutil.Info("yyyyyy zero txnOffset",
-			zap.String("txn", hex.EncodeToString(ls.table.db.op.Txn().ID)),
-			zap.Bool("isSnapOp", ls.table.db.op.IsSnapOp()),
-			zap.String("entries", stringifySlice(writes[len(writes)-2:], func(a any) string {
-				e := a.(Entry)
-				batstr := "nil"
-				if e.bat != nil {
-					batstr = common.MoBatchToString(e.bat, 3)
-				}
-				return e.String() + " " + batstr
-			})))
-	}
-}
-
 func checkTxnLastInsertRow(ls *LocalDisttaeDataSource, writes []Entry, cursor int, outBatch *batch.Batch) {
-	if len(writes) > 400 && ls.table.accountId == 0 && ls.table.tableName == "mo_increment_columns" && writes[len(writes)-1].typ == INSERT && writes[len(writes)-1].tableId == ls.table.tableId {
+	injected, writesT := objectio.Debug19357Injected()
+	if injected && int64(len(writes)) > writesT && len(outBatch.Vecs) == 3 && ls.table.accountId == 0 && ls.table.tableName == "mo_increment_columns" && writes[len(writes)-1].typ == INSERT && writes[len(writes)-1].tableId == ls.table.tableId {
+		outLen := outBatch.Vecs[0].Length()
+		var slim *batch.Batch
+		if outLen > 0 {
+			start := outLen - 3
+			if start < 0 {
+				start = 0
+			}
+			slim, _ = outBatch.Window(start, outLen)
+		}
+
 		logutil.Info("yyyyyy checkTxnLastInsertRow",
 			zap.String("txn", hex.EncodeToString(ls.table.db.op.Txn().ID)),
 			zap.Int("txnOffset", ls.txnOffset),
 			zap.Int("cursor", cursor),
 			zap.Int("writes", len(writes)),
 			zap.Bool("isSnapOp", ls.table.db.op.IsSnapOp()),
-			zap.String("entries", stringifySlice(writes[len(writes)-2:], func(a any) string {
+			zap.String("entries", stringifySlice(writes[len(writes)-1:], func(a any) string {
 				e := a.(Entry)
 				batstr := "nil"
 				if e.bat != nil {
 					batstr = common.MoBatchToString(e.bat, 3)
 				}
 				return e.String() + " " + batstr
 			})),
-			zap.String("outBatch", common.MoBatchToString(outBatch, 3)),
+			zap.String("outBatch", common.MoBatchToString(slim, 3)),
 		)
 	}
 }
@@ -488,13 +483,20 @@ func (ls *LocalDisttaeDataSource) filterInMemUnCommittedInserts(
 		retainedRowIds = vector.MustFixedColWithTypeCheck[objectio.Rowid](entry.bat.Vecs[0])
 		offsets := engine_util.RowIdsToOffset(retainedRowIds, int64(0)).([]int64)
 
+		offsetLen := len(offsets)
+		badOffsetStart := offsetLen > 0 && offsets[0] > 0
+
 		b := retainedRowIds[0].BorrowBlockID()
 		sels, err := ls.ApplyTombstones(
 			ls.ctx, b, offsets, engine.Policy_CheckUnCommittedOnly)
 		if err != nil {
 			return err
 		}
 
+		if (len(sels) < offsetLen || badOffsetStart) && ls.table.accountId == 0 && ls.table.tableName == "mo_increment_columns" {
+			logutil.Info("Shrink retainedRowIds", zap.Any("sels", sels), zap.Any("offsetsLen", offsetLen), zap.Bool("badOffsetStart", badOffsetStart), zap.Int("wsCursor", ls.wsCursor), zap.Int("txnOffset", ls.txnOffset))
+		}
+
 		if len(sels) == 0 {
 			continue
 		}

pkg/vm/engine/disttae/local_disttae_datasource_test.go

@@ -21,6 +21,7 @@ import (
 
 	"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"
@@ -30,6 +31,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
+	"github.com/matrixorigin/matrixone/pkg/util/fault"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/logtailreplay"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/engine_util"
 )
@@ -161,8 +163,18 @@ func TestXxx1(t *testing.T) {
 			},
 		},
 	}
+	fault.Enable()
+	defer fault.Disable()
+	fault.AddFaultPoint(context.Background(), objectio.FJ_Debug19357, ":::", "echo", 20, "")
+	t.Log(objectio.Debug19357Injected())
 	writes := make([]Entry, 200)
 	writes = append(writes, Entry{typ: INSERT}, Entry{typ: INSERT, bat: batch.EmptyBatch})
-	checkTxnLastInsertRow(ls, writes, 42, batch.EmptyBatch)
-	checkTxnOffsetZero(ls, writes)
+	outBatch := batch.NewWithSize(3)
+	m := mpool.MustNewZero()
+	for i := 0; i < 3; i++ {
+		outBatch.Vecs[i] = vector.NewVec(types.T_int32.ToType())
+		vector.AppendFixed[int32](outBatch.Vecs[i], int32(i), false, m)
+	}
+	checkTxnLastInsertRow(ls, writes, 42, outBatch)
+
 }

pkg/vm/engine/tae/logtail/snapshot.go

@@ -495,7 +495,7 @@ func (sm *SnapshotMeta) updateTableInfo(
 				continue
 			}
 			if _, ok := sm.aobjDelTsMap[commitTs]; ok {
-				logutil.Infof("yyyy skip table %v @ %v", pk.ErrString(nil), commitTs.ToString())
+				logutil.Infof("snapshotMeta skip table %v @ %v", pk.ErrString(nil), commitTs.ToString())
 				continue
 			}
 			deleteRows = append(deleteRows, tombstone{