add summary for txnBlock(1.2-dev only)

pkg/util/metric/v2/dashboard/grafana_dashboard_logtail.go

@@ -70,6 +70,20 @@ func (c *DashboardCreator) initLogtailCollectRow() dashboard.Option {
 			4,
 			axis.Unit("s"),
 			axis.Min(0)),
+
+		c.getHistogram(
+			"pull scan row count",
+			c.getMetricWithFilter("mo_logtail_pull_scan_txn_count_bucket", `type="scan-row"`),
+			[]float64{0.50, 0.7, 0.8, 0.90},
+			6,
+			axis.Min(0)),
+
+		c.getHistogram(
+			"pull skip blk count",
+			c.getMetricWithFilter("mo_logtail_pull_scan_txn_count_bucket", `type="skip-blk"`),
+			[]float64{0.50, 0.7, 0.8, 0.90},
+			6,
+			axis.Min(0)),
 	)
 }
 

pkg/util/metric/v2/logtail.go

@@ -119,6 +119,18 @@ var (
 	LogtailSendLatencyHistogram = logTailSendDurationHistogram.WithLabelValues("latency")
 	LogtailSendNetworkHistogram = logTailSendDurationHistogram.WithLabelValues("network")
 
+	LogtailPullScanTxnCountHistogram = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "mo",
+			Subsystem: "logtail",
+			Name:      "pull_scan_txn_count",
+			Help:      "Bucketed histogram of pull scan txn count.",
+			Buckets:   prometheus.ExponentialBuckets(1, 3.0, 20),
+		}, []string{"type"})
+
+	LogTailPullScanSkipBlkCountHistogram = LogtailPullScanTxnCountHistogram.WithLabelValues("skip-blk")
+	LogTailPullScanScanRowCountHistogram = LogtailPullScanTxnCountHistogram.WithLabelValues("scan-row")
+
 	LogTailLoadCheckpointDurationHistogram = prometheus.NewHistogram(
 		prometheus.HistogramOpts{
 			Namespace: "mo",

pkg/util/metric/v2/metrics.go

@@ -109,6 +109,7 @@ func initLogtailMetrics() {
 	registry.MustRegister(logTailSendDurationHistogram)
 	registry.MustRegister(LogTailLoadCheckpointDurationHistogram)
 
+	registry.MustRegister(LogtailPullScanTxnCountHistogram)
 	registry.MustRegister(LogTailPushCollectionDurationHistogram)
 	registry.MustRegister(LogTailPullCollectionPhase1DurationHistogram)
 	registry.MustRegister(LogTailPullCollectionPhase2DurationHistogram)

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

@@ -199,7 +199,7 @@ func (mgr *Manager) GCByTS(ctx context.Context, ts types.TS) {
 	}
 	mgr.truncated = ts
 	cnt := mgr.table.TruncateByTimeStamp(ts)
-	logutil.Info("[logtail] GC", zap.String("ts", ts.ToString()), zap.Int("deleted", cnt))
+	logutil.Info("[logtail] GC", zap.String("ts", ts.ToString()), zap.Int("deleted-blk", cnt), zap.Int("remaining-blk", mgr.table.BlockCount()))
 }
 
 func (mgr *Manager) TryCompactTable() {

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

@@ -16,6 +16,7 @@ package logtail
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/container/types"
+	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/txnif"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/model"
 )
@@ -77,13 +78,28 @@ func (r *Reader) GetDirtyByTable(
 	dbID, id uint64,
 ) (tree *model.TableTree) {
 	tree = model.NewTableTree(dbID, id)
+	var rowScan, blkSkip int
 	op := func(row RowT) (moveOn bool) {
+		rowScan++
 		if memo := row.GetMemo(); memo.HasTableDataChanges(id) {
 			tree.Merge(memo.GetDirtyTableByID(id))
 		}
 		return true
 	}
-	r.table.ForeachRowInBetween(r.from, r.to, nil, op)
+	skipFn := func(blk BlockT) bool {
+		summary := blk.summary.Load()
+		if summary == nil {
+			return false
+		}
+		_, exist := summary.tids[id]
+		if !exist {
+			blkSkip++
+		}
+		return !exist
+	}
+	r.table.ForeachRowInBetween(r.from, r.to, skipFn, op)
+	v2.LogTailPullScanSkipBlkCountHistogram.Observe(float64(blkSkip))
+	v2.LogTailPullScanScanRowCountHistogram.Observe(float64(rowScan))
 	return
 }
 

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

@@ -66,8 +66,8 @@ func (txn *smallTxn) GetLSN() uint64 {
 
 type summary struct {
 	hasCatalogChanges bool
+	tids              map[uint64]struct{}
 	// TODO
-	// table ids
 	// maxLsn
 }
 
@@ -207,10 +207,14 @@ func (blk *txnBlock) Close() {
 
 func (blk *txnBlock) trySumary() {
 	summary := new(summary)
+	summary.tids = make(map[uint64]struct{}, 8)
 	for _, row := range blk.rows {
-		if row.GetMemo().HasCatalogChanges() {
+		memo := row.GetMemo()
+		if !summary.hasCatalogChanges && memo.HasCatalogChanges() {
 			summary.hasCatalogChanges = true
-			break
+		}
+		for k := range memo.Tree.Tables {
+			summary.tids[k] = struct{}{}
 		}
 	}
 	blk.summary.CompareAndSwap(nil, summary)

pkg/vm/engine/tae/options/types.go

@@ -53,7 +53,7 @@ const (
 	DefaultIOWorkers    = int(16)
 	DefaultAsyncWorkers = int(16)
 
-	DefaultLogtailTxnPageSize = 100
+	DefaultLogtailTxnPageSize = 256
 
 	DefaultLogstoreType = LogstoreBatchStore
 )