Skip to content

Commit

Permalink
Merge branch '1.2-dev' into issue16357-1.2
Browse files Browse the repository at this point in the history
  • Loading branch information
@mergify
mergify[bot] authored May 24, 2024
2 parents 7f296d2 + 3e58ca6 commit c1728d0
Show file tree
Hide file tree
Showing 20 changed files with 441 additions and 1,741 deletions.
106 changes: 45 additions & 61 deletions pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,16 +38,15 @@ func (c *DashboardCreator) initTxnDashboard() error {
c.initTxnLockWaitersRow(),
c.initTxnStatementDurationRow(),
c.initTxnStatementsCountRow(),
c.initTxnTableRangesRow(),
c.initTxnCheckPKDupRow(),
c.initTxnReaderDurationRow(),
c.initTxnMpoolRow(),
c.initTxnOnPrepareWALRow(),
c.initTxnBeforeCommitRow(),
c.initTxnDequeuePreparedRow(),
c.initTxnRangesLoadedObjectMetaRow(),
c.initFastRangesRow(),
c.initRangesRow(),
c.initTxnTableRangesRow(),
c.initTxnRangesSelectivityRow(),
c.initTxnRangesCountRow(),
c.initTxnShowAccountsRow(),
c.initCNCommittedObjectQuantityRow(),
)...)
Expand Down Expand Up @@ -75,66 +74,62 @@ func (c *DashboardCreator) initCNCommittedObjectQuantityRow() dashboard.Option {
)
}

func (c *DashboardCreator) initRangesRow() dashboard.Option {
func (c *DashboardCreator) initTxnTableRangesRow() dashboard.Option {
return dashboard.Row(
"Txn Ranges Selectivity",
"Txn Table Ranges Duration",
c.getHistogram(
"ranges block selectivity",
c.getMetricWithFilter("mo_txn_ranges_selectivity_percentage_bucket", `type="block_selectivity"`),
[]float64{0.50, 0.8, 0.90, 0.99},
6,
axis.Unit(""),
axis.Min(0)),

c.getHistogram(
"ranges result len",
c.getMetricWithFilter("mo_txn_ranges_duration_size_bucket", `type="ranges_len"`),
"Txn table ranges duration",
c.getMetricWithFilter(`mo_txn_ranges_duration_seconds_bucket`, ``),
[]float64{0.50, 0.8, 0.90, 0.99},
6,
axis.Unit(""),
12,
axis.Unit("s"),
axis.Min(0)),
)
}

func (c *DashboardCreator) initFastRangesRow() dashboard.Option {
func (c *DashboardCreator) initTxnRangesSelectivityRow() dashboard.Option {
return dashboard.Row(
"Txn Fast Ranges Selectivity",
c.getHistogram(
"fast ranges block selectivity",
c.getMetricWithFilter("mo_txn_ranges_selectivity_percentage_bucket", `type="fast_block_selectivity"`),
[]float64{0.50, 0.8, 0.90, 0.99},
4,
axis.Unit(""),
axis.Min(0)),

c.getHistogram(
"fast ranges zone map selectivity",
c.getMetricWithFilter("mo_txn_ranges_selectivity_percentage_bucket", `type="fast_zm_selectivity"`),
[]float64{0.50, 0.8, 0.90, 0.99},
4,
axis.Unit(""),
axis.Min(0)),

c.getHistogram(
"fast ranges result len",
c.getMetricWithFilter("mo_txn_ranges_duration_size_bucket", `type="fast_ranges_len"`),
"Ranges Selectivity",
c.getMultiHistogram(
[]string{
c.getMetricWithFilter(`mo_txn_ranges_selectivity_percentage_bucket`, `type="slow_path_block_selectivity"`),
c.getMetricWithFilter(`mo_txn_ranges_selectivity_percentage_bucket`, `type="fast_path_block_selectivity"`),
c.getMetricWithFilter(`mo_txn_ranges_selectivity_percentage_bucket`, `type="fast_path_obj_sort_key_zm_selectivity"`),
c.getMetricWithFilter(`mo_txn_ranges_selectivity_percentage_bucket`, `type="fast_path_obj_column_zm_selectivity"`),
c.getMetricWithFilter(`mo_txn_ranges_selectivity_percentage_bucket`, `type="fast_path_blk_column_zm_selectivity"`),
},
[]string{
"slow_path_block_selectivity",
"fast_path_block_selectivity",
"fast_path_obj_sort_key_zm_selectivity",
"fast_path_obj_column_zm_selectivity",
"fast_path_blk_column_zm_selectivity",
},
[]float64{0.50, 0.8, 0.90, 0.99},
4,
axis.Unit(""),
axis.Min(0)),
[]float32{3, 3, 3, 3},
axis.Min(0))...,
)
}

func (c *DashboardCreator) initTxnTableRangesRow() dashboard.Option {
func (c *DashboardCreator) initTxnRangesCountRow() dashboard.Option {
return dashboard.Row(
"Txn table ranges",
c.getHistogram(
"Txn table ranges duration",
c.getMetricWithFilter(`mo_txn_ranges_duration_seconds_bucket`, ``),
"Ranges Count",
c.getMultiHistogram(
[]string{
c.getMetricWithFilter(`mo_txn_ranges_selected_block_cnt_total_bucket`, `type="slow_path_selected_block_cnt"`),
c.getMetricWithFilter(`mo_txn_ranges_selected_block_cnt_total_bucket`, `type="fast_path_selected_block_cnt"`),
c.getMetricWithFilter(`mo_txn_ranges_selected_block_cnt_total_bucket`, `type="fast_path_load_obj_cnt"`),
c.getMetricWithFilter(`mo_txn_ranges_selected_block_cnt_total_bucket`, `type="slow_path_load_obj_cnt"`),
},
[]string{
"slow_path_selected_block_cnt",
"fast_path_selected_block_cnt",
"fast_path_load_obj_cnt",
"slow_path_load_obj_cnt",
},
[]float64{0.50, 0.8, 0.90, 0.99},
12,
axis.Unit("s"),
axis.Min(0)),
[]float32{3, 3, 3, 3},
axis.Min(0))...,
)
}

Expand Down Expand Up @@ -403,17 +398,6 @@ func (c *DashboardCreator) initTxnStatementsCountRow() dashboard.Option {
)
}

func (c *DashboardCreator) initTxnRangesLoadedObjectMetaRow() dashboard.Option {
return dashboard.Row(
"Txn Ranges Loaded Object Meta",
c.withGraph(
"Txn Ranges Loaded Object Meta",
12,
`sum(increase(`+c.getMetricWithFilter("mo_txn_ranges_loaded_object_meta_total", "")+`[$interval])) by (`+c.by+`, type)`,
"{{ "+c.by+"-type }}"),
)
}

func (c *DashboardCreator) initTxnShowAccountsRow() dashboard.Option {
return dashboard.Row(
"Show Accounts Duration",
Expand Down
3 changes: 1 addition & 2 deletions pkg/util/metric/v2/metrics.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -130,12 +130,11 @@ func initTxnMetrics() {
registry.MustRegister(TxnTableRangeDurationHistogram)
registry.MustRegister(TxnCheckPKDupDurationHistogram)
registry.MustRegister(TxnLockWaitersTotalHistogram)
registry.MustRegister(txnTableRangeSizeHistogram)
registry.MustRegister(txnTableRangeTotalHistogram)
registry.MustRegister(txnMpoolDurationHistogram)
registry.MustRegister(TxnUnlockTableTotalHistogram)
registry.MustRegister(txnReaderDurationHistogram)

registry.MustRegister(TxnRangesLoadedObjectMetaTotalCounter)
registry.MustRegister(txnCNCommittedLocationQuantityGauge)

registry.MustRegister(txnRangesSelectivityHistogram)
Expand Down
32 changes: 14 additions & 18 deletions pkg/util/metric/v2/txn.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,14 +72,6 @@ var (
TxnLockTotalCounter = txnLockCounter.WithLabelValues("total")
TxnLocalLockTotalCounter = txnLockCounter.WithLabelValues("local")
TxnRemoteLockTotalCounter = txnLockCounter.WithLabelValues("remote")

TxnRangesLoadedObjectMetaTotalCounter = prometheus.NewCounter(
prometheus.CounterOpts{
Namespace: "mo",
Subsystem: "txn",
Name: "ranges_loaded_object_meta_total",
Help: "Total number of ranges loaded object meta.",
})
)

var (
Expand Down Expand Up @@ -247,17 +239,19 @@ var (
Buckets: getDurationBuckets(),
})

txnTableRangeSizeHistogram = prometheus.NewHistogramVec(
txnTableRangeTotalHistogram = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Namespace: "mo",
Subsystem: "txn",
Name: "ranges_duration_size",
Help: "Bucketed histogram of txn table ranges size.",
Buckets: prometheus.ExponentialBuckets(1, 2.0, 20),
Name: "ranges_selected_block_cnt_total",
Help: "Bucketed histogram of txn table ranges selected block cnt.",
Buckets: prometheus.ExponentialBuckets(1, 2.0, 15),
}, []string{"type"})

TxnRangeSizeHistogram = txnTableRangeSizeHistogram.WithLabelValues("ranges_len")
TxnFastRangeSizeHistogram = txnTableRangeSizeHistogram.WithLabelValues("fast_ranges_len")
TxnRangesSlowPathSelectedBlockCntHistogram = txnTableRangeTotalHistogram.WithLabelValues("slow_path_selected_block_cnt")
TxnRangesFastPathSelectedBlockCntHistogram = txnTableRangeTotalHistogram.WithLabelValues("fast_path_selected_block_cnt")
TxnRangesFastPathLoadObjCntHistogram = txnTableRangeTotalHistogram.WithLabelValues("fast_path_load_obj_cnt")
TxnRangesSlowPathLoadObjCntHistogram = txnTableRangeTotalHistogram.WithLabelValues("slow_path_load_obj_cnt")

txnTNSideDurationHistogram = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Expand Down Expand Up @@ -307,9 +301,11 @@ var (
Subsystem: "txn",
Name: "ranges_selectivity_percentage",
Help: "Bucketed histogram of fast ranges selectivity percentage.",
Buckets: prometheus.LinearBuckets(0, 0.05, 21),
Buckets: prometheus.ExponentialBucketsRange(0.001, 1, 21),
}, []string{"type"})
TxnRangesBlockSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("block_selectivity")
TxnFastRangesBlockSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("fast_block_selectivity")
TxnFastRangesZMapSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("fast_zm_selectivity")
TxnRangesSlowPathBlockSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("slow_path_block_selectivity")
TxnRangesFastPathBlkTotalSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("fast_path_block_selectivity")
TxnRangesFastPathObjSortKeyZMapSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("fast_path_obj_sort_key_zm_selectivity")
TxnRangesFastPathObjColumnZMapSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("fast_path_obj_column_zm_selectivity")
TxnRangesFastPathBlkColumnZMapSelectivityHistogram = txnRangesSelectivityHistogram.WithLabelValues("fast_path_blk_column_zm_selectivity")
)
34 changes: 34 additions & 0 deletions pkg/vm/engine/disttae/filter.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
"github.com/matrixorigin/matrixone/pkg/pb/plan"
"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
"github.com/matrixorigin/matrixone/pkg/util"
v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/logtailreplay"
"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/index"
"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/options"
Expand Down Expand Up @@ -1021,15 +1022,42 @@ func ExecuteBlockFilter(
fs fileservice.FileService,
proc *process.Process,
) (err error) {
var (
totalBlocks float64
loadHit float64
objFilterTotal, objFilterHit float64
blkFilterTotal, blkFilterHit float64
fastFilterTotal, fastFilterHit float64
)

defer func() {
v2.TxnRangesFastPathLoadObjCntHistogram.Observe(loadHit)
v2.TxnRangesFastPathSelectedBlockCntHistogram.Observe(float64(outBlocks.Len() - 1))
if fastFilterTotal > 0 {
v2.TxnRangesFastPathObjSortKeyZMapSelectivityHistogram.Observe(fastFilterHit / fastFilterTotal)
}
if objFilterTotal > 0 {
v2.TxnRangesFastPathObjColumnZMapSelectivityHistogram.Observe(objFilterHit / objFilterTotal)
}
if blkFilterTotal > 0 {
v2.TxnRangesFastPathBlkColumnZMapSelectivityHistogram.Observe(blkFilterHit / blkFilterTotal)
}
if totalBlocks > 0 {
v2.TxnRangesFastPathBlkTotalSelectivityHistogram.Observe(float64(outBlocks.Len()-1) / totalBlocks)
}
}()

hasDeletes := len(dirtyBlocks) > 0
err = ForeachSnapshotObjects(
snapshotTS,
func(obj logtailreplay.ObjectInfo, isCommitted bool) (err2 error) {
var ok bool
objStats := obj.ObjectStats
totalBlocks += float64(objStats.BlkCnt())
if fastFilterOp != nil {
fastFilterTotal++
if ok, err2 = fastFilterOp(objStats); err2 != nil || !ok {
fastFilterHit++
return
}
}
Expand All @@ -1038,14 +1066,17 @@ func ExecuteBlockFilter(
bf objectio.BloomFilter
)
if loadOp != nil {
loadHit++
if meta, bf, err2 = loadOp(
proc.Ctx, objStats, meta, bf,
); err2 != nil {
return
}
}
if objectFilterOp != nil {
objFilterTotal++
if ok, err2 = objectFilterOp(meta, bf); err2 != nil || !ok {
objFilterHit++
return
}
}
Expand Down Expand Up @@ -1078,6 +1109,7 @@ func ExecuteBlockFilter(
for ; pos < blockCnt; pos++ {
var blkMeta objectio.BlockObject
if dataMeta != nil && blockFilterOp != nil {
blkFilterTotal++
var (
quickBreak, ok2 bool
)
Expand All @@ -1088,10 +1120,12 @@ func ExecuteBlockFilter(
}
// skip the following block checks
if quickBreak {
blkFilterHit++
break
}
// skip this block
if !ok2 {
blkFilterHit++
continue
}
}
Expand Down
72 changes: 5 additions & 67 deletions pkg/vm/engine/disttae/reader.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -133,79 +133,17 @@ func (mixin *withFilterMixin) getReadFilter(proc *process.Process, blkCnt int) (
return
}
pk := mixin.tableDef.Pkey
if pk == nil {
if mixin.filterState.expr == nil || pk == nil {
mixin.filterState.evaluated = true
mixin.filterState.filter = nil
return
}
if pk.CompPkeyCol == nil {
return mixin.getNonCompositPKFilter(proc, blkCnt)
}
return mixin.getCompositPKFilter(proc, blkCnt)
}

func (mixin *withFilterMixin) getCompositPKFilter(proc *process.Process, blkCnt int) (
filter blockio.ReadFilter,
) {
// if no primary key is included in the columns or no filter expr is given,
// no filter is needed
if len(mixin.columns.compPKPositions) == 0 || mixin.filterState.expr == nil {
mixin.filterState.evaluated = true
mixin.filterState.filter = nil
return
}

// evaluate
pkNames := mixin.tableDef.Pkey.Names
pkVals := make([]*plan.Literal, len(pkNames))
ok, hasNull := getCompositPKVals(mixin.filterState.expr, pkNames, pkVals, proc)

if !ok || pkVals[0] == nil {
mixin.filterState.evaluated = true
mixin.filterState.filter = nil
mixin.filterState.hasNull = hasNull
return
}
cnt := getValidCompositePKCnt(pkVals)
pkVals = pkVals[:cnt]

filterFuncs := make([]func(*vector.Vector, []int32, *[]int32), len(pkVals))
for i := range filterFuncs {
filterFuncs[i] = getCompositeFilterFuncByExpr(pkVals[i], i == 0)
}

filter = func(vecs []*vector.Vector) []int32 {
var (
inputSels []int32
)
for i := range filterFuncs {
vec := vecs[i]
mixin.sels = mixin.sels[:0]
filterFuncs[i](vec, inputSels, &mixin.sels)
if len(mixin.sels) == 0 {
break
}
inputSels = mixin.sels
}
// logutil.Debugf("%s: %d/%d", mixin.tableDef.Name, len(res), vecs[0].Length())

return mixin.sels
}

mixin.filterState.evaluated = true
mixin.filterState.filter = filter
mixin.filterState.seqnums = make([]uint16, 0, len(mixin.columns.compPKPositions))
mixin.filterState.colTypes = make([]types.Type, 0, len(mixin.columns.compPKPositions))
for _, pos := range mixin.columns.compPKPositions {
mixin.filterState.seqnums = append(mixin.filterState.seqnums, mixin.columns.seqnums[pos])
mixin.filterState.colTypes = append(mixin.filterState.colTypes, mixin.columns.colTypes[pos])
}
// records how many blks one reader needs to read when having filter
objectio.BlkReadStats.BlksByReaderStats.Record(1, blkCnt)
return
return mixin.getNonCompositPKFilter(proc, blkCnt)
}

func (mixin *withFilterMixin) getNonCompositPKFilter(proc *process.Process, blkCnt int) blockio.ReadFilter {
func (mixin *withFilterMixin) getNonCompositPKFilter(
proc *process.Process, blkCnt int,
) blockio.ReadFilter {
// if no primary key is included in the columns or no filter expr is given,
// no filter is needed
if mixin.columns.pkPos == -1 || mixin.filterState.expr == nil {
Expand Down
Loading

0 comments on commit c1728d0

Please sign in to comment.