metrics

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

@@ -43,7 +43,7 @@ func (c *DashboardCreator) initTxnDashboard() error {
 			c.initTxnMpoolRow(),
 			c.initTxnOnPrepareWALRow(),
 			c.initTxnBeforeCommitRow(),
-			c.initTxnDequeuePreparedRow(),
+			c.initTxnTNDeduplicateDurationRow(),
 			c.initTxnTableRangesRow(),
 			c.initTxnRangesSelectivityRow(),
 			c.initTxnRangesCountRow(),
@@ -292,16 +292,22 @@ func (c *DashboardCreator) initTxnOnPrepareWALRow() dashboard.Option {
 	)
 }
 
-func (c *DashboardCreator) initTxnDequeuePreparedRow() dashboard.Option {
+func (c *DashboardCreator) initTxnTNDeduplicateDurationRow() dashboard.Option {
 	return dashboard.Row(
-		"txn dequeue prepared duration",
-		c.getHistogram(
-			"txn dequeue prepared duration",
-			c.getMetricWithFilter("mo_txn_tn_side_duration_seconds_bucket", `step="dequeue_prepared"`),
-			[]float64{0.50, 0.8, 0.90, 0.99},
-			12,
+		"Txn TN Deduplication Duration",
+		c.getMultiHistogram(
+			[]string{
+				c.getMetricWithFilter("mo_txn_tn_deduplicate_duration_seconds_bucket", `type="append_deduplicate"`),
+				c.getMetricWithFilter("mo_txn_tn_deduplicate_duration_seconds_bucket", `type="prePrepare_deduplicate"`),
+			},
+			[]string{
+				"append_deduplicate",
+				"prePrepare_deduplicate",
+			},
+			[]float64{0.80, 0.90, 0.95, 0.99},
+			[]float32{3, 3, 3, 3},
 			axis.Unit("s"),
-			axis.Min(0)),
+			axis.Min(0))...,
 	)
 }
 

pkg/util/metric/v2/metrics.go

@@ -140,6 +140,7 @@ func initTxnMetrics() {
 	registry.MustRegister(txnCNCommittedLocationQuantityGauge)
 
 	registry.MustRegister(txnRangesSelectivityHistogram)
+	registry.MustRegister(txnTNDeduplicateDurationHistogram)
 }
 
 func initRPCMetrics() {

pkg/util/metric/v2/txn.go

@@ -269,8 +269,19 @@ var (
 	TxnPreparedWaitDurationHistogram   = txnTNSideDurationHistogram.WithLabelValues("5-PreparedWait")
 	TxnPreparedDurationHistogram       = txnTNSideDurationHistogram.WithLabelValues("6-Prepared")
 
-	TxnDequeuePreparedDurationHistogram = txnTNSideDurationHistogram.WithLabelValues("dequeue_prepared")
-	TxnBeforeCommitDurationHistogram    = txnTNSideDurationHistogram.WithLabelValues("before_txn_commit")
+	TxnBeforeCommitDurationHistogram = txnTNSideDurationHistogram.WithLabelValues("before_txn_commit")
+
+	txnTNDeduplicateDurationHistogram = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "mo",
+			Subsystem: "txn",
+			Name:      "tn_deduplicate_duration_seconds",
+			Help:      "Bucketed histogram of txn duration on tn side",
+			Buckets:   getDurationBuckets(),
+		}, []string{"type"})
+
+	TxnTNAppendDeduplicateDurationHistogram     = txnTNDeduplicateDurationHistogram.WithLabelValues("append_deduplicate")
+	TxnTNPrePrepareDeduplicateDurationHistogram = txnTNDeduplicateDurationHistogram.WithLabelValues("prePrepare_deduplicate")
 
 	txnMpoolDurationHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{

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

@@ -22,7 +22,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
 	"github.com/panjf2000/ants/v2"
 	"go.uber.org/zap"
 
@@ -559,7 +558,6 @@ func (mgr *TxnManager) dequeuePrepared(items ...any) {
 		store.TriggerTrace(txnif.TracePrepared)
 		mgr.workers.Submit(func() {
 			//Notice that WaitPrepared do nothing when op is OpRollback
-			t0 := time.Now()
 			if err := op.Txn.WaitPrepared(op.ctx); err != nil {
 				// v0.6 TODO: Error handling
 				panic(err)
@@ -570,8 +568,6 @@ func (mgr *TxnManager) dequeuePrepared(items ...any) {
 			} else {
 				mgr.on1PCPrepared(op)
 			}
-			dequeuePreparedDuration := time.Since(t0)
-			v2.TxnDequeuePreparedDurationHistogram.Observe(dequeuePreparedDuration.Seconds())
 		})
 	}
 	common.DoIfDebugEnabled(func() {

pkg/vm/engine/tae/txn/txnimpl/table.go

@@ -21,20 +21,17 @@ import (
 	"runtime/trace"
 	"time"
 
-	"github.com/matrixorigin/matrixone/pkg/fileservice"
-	"github.com/matrixorigin/matrixone/pkg/util"
-	"go.uber.org/zap"
-
-	"github.com/matrixorigin/matrixone/pkg/perfcounter"
-
 	"github.com/RoaringBitmap/roaring"
-
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/moprobe"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	apipb "github.com/matrixorigin/matrixone/pkg/pb/api"
+	"github.com/matrixorigin/matrixone/pkg/perfcounter"
+	"github.com/matrixorigin/matrixone/pkg/util"
+	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/blockio"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
@@ -45,6 +42,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/model"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/tables/updates"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/wal"
+	"go.uber.org/zap"
 )
 
 var (
@@ -589,7 +587,9 @@ func (tbl *txnTable) AddDeleteNode(id *common.ID, node txnif.DeleteNode) error {
 }
 
 func (tbl *txnTable) Append(ctx context.Context, data *containers.Batch) (err error) {
+	var dedupDur float64
 	if tbl.schema.HasPK() && !tbl.schema.IsSecondaryIndexTable() {
+		now := time.Now()
 		dedupType := tbl.store.txn.GetDedupType()
 		if dedupType == txnif.FullDedup {
 			//do PK deduplication check against txn's work space.
@@ -616,11 +616,19 @@ func (tbl *txnTable) Append(ctx context.Context, data *containers.Batch) (err er
 				return
 			}
 		}
+		dedupDur += time.Since(now).Seconds()
 	}
+
 	if tbl.tableSpace == nil {
 		tbl.tableSpace = newTableSpace(tbl)
 	}
-	return tbl.tableSpace.Append(data)
+
+	dur, err := tbl.tableSpace.Append(data)
+	dedupDur += dur
+
+	v2.TxnTNAppendDeduplicateDurationHistogram.Observe(dedupDur)
+
+	return err
 }
 func (tbl *txnTable) AddObjsWithMetaLoc(ctx context.Context, stats containers.Vector) (err error) {
 	return stats.Foreach(func(v any, isNull bool, row int) error {

pkg/vm/engine/tae/txn/txnimpl/table_space.go

@@ -17,6 +17,7 @@ package txnimpl
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
@@ -280,7 +281,7 @@ func (space *tableSpace) CloseAppends() {
 }
 
 // Append appends batch of data into anode.
-func (space *tableSpace) Append(data *containers.Batch) (err error) {
+func (space *tableSpace) Append(data *containers.Batch) (dur float64, err error) {
 	if space.appendable == nil {
 		space.registerANode()
 	}
@@ -296,6 +297,7 @@ func (space *tableSpace) Append(data *containers.Batch) (err error) {
 		}
 		dedupType := space.table.store.txn.GetDedupType()
 		if schema.HasPK() && dedupType == txnif.FullDedup {
+			now := time.Now()
 			if err = space.index.BatchInsert(
 				data.Attrs[schema.GetSingleSortKeyIdx()],
 				data.Vecs[schema.GetSingleSortKeyIdx()],
@@ -305,6 +307,7 @@ func (space *tableSpace) Append(data *containers.Batch) (err error) {
 				false); err != nil {
 				break
 			}
+			dur += time.Since(now).Seconds()
 		}
 		offset += appended
 		space.rows += appended

pkg/vm/engine/tae/txn/txnimpl/txndb.go

@@ -16,6 +16,7 @@ package txnimpl
 
 import (
 	"context"
+	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
 	"sync"
 	"time"
 
@@ -464,11 +465,15 @@ func (db *txnDB) PrePrepare(ctx context.Context) (err error) {
 			return
 		}
 	}
+
+	now := time.Now()
 	for _, table := range db.tables {
 		if err = table.PrePrepareDedup(ctx); err != nil {
 			return
 		}
 	}
+	v2.TxnTNPrePrepareDeduplicateDurationHistogram.Observe(time.Since(now).Seconds())
+
 	for _, table := range db.tables {
 		if err = table.PrePrepare(); err != nil {
 			return