Repository: tikv/client-go
Pull request #1606 (2025-03-12 07:04:11)

execdetails: export scheduler write details
Export scheduler write details for write requests.

TiDB slow query example:

```diff
@@ -380,7 +384,8 @@ func TestRuntimeStatsWithCommit(t *testing.T) {
                "block: {cache_hit_count: 1, read_count: 1, read_byte: 100 Bytes, read_time: 20ms}}}, " +
                "write_detail: {store_batch_wait: 10µs, propose_send_wait: 20µs, persist_log: {total: 30µs, write_leader_wait: 40µs, " +
                "sync_log: 45µs, write_memtable: 50µs}, commit_log: 60µs, apply_batch_wait: 70µs, apply: {total:80µs, mutex_lock: 90µs, " +
-               "write_leader_wait: 100µs, write_wal: 101µs, write_memtable: 102µs}}}, resolve_lock: 1s, region_num:5, write_keys:3" +
+               "write_leader_wait: 100µs, write_wal: 101µs, write_memtable: 102µs}, scheduler: {process: 104µs, latch_wait: 103µs, " +
+               "pessimistic_lock_wait: 106µs, throttle: 105µs}}}, resolve_lock: 1s, region_num:5, write_keys:3" +
                ", write_byte:66, txn_retry:2}"
        require.Equal(t, expect, stats.String())
 ```


Cc https://github.com/tikv/tikv/issues/18274
Cc https://github.com/pingcap/tidb/pull/60030

ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [overvenus](https://github.com/overvenus), **ensuring that each of them provides their approval before proceeding**. For more information see [the Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=tikv%2Fclient-go).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/tikv/client-go/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["overvenus"]} -->

Patch:
diff --git a/oracle/oracles/local.go b/oracle/oracles/local.go
index a980bbd..e5aa21f 100644
--- a/oracle/oracles/local.go
+++ b/oracle/oracles/local.go
@@ -36,11 +36,9 @@ package oracles
 
 import (
 	"context"
-	"math"
 	"sync"
 	"time"
 
-	"github.com/pingcap/errors"
 	"github.com/tikv/client-go/v2/oracle"
 )
 
@@ -152,22 +150,6 @@ func (l *localOracle) GetExternalTimestamp(ctx context.Context) (uint64, error)
 }
 
 func (l *localOracle) ValidateReadTS(ctx context.Context, readTS uint64, isStaleRead bool, opt *oracle.Option) error {
-	if readTS == math.MaxUint64 {
-		if isStaleRead {
-			return oracle.ErrLatestStaleRead{}
-		}
-		return nil
-	}
-
-	currentTS, err := l.GetTimestamp(ctx, opt)
-	if err != nil {
-		return errors.Errorf("fail to validate read timestamp: %v", err)
-	}
-	if currentTS < readTS {
-		return oracle.ErrFutureTSRead{
-			ReadTS:    readTS,
-			CurrentTS: currentTS,
-		}
-	}
+	// local oracle is not supposed to be used
 	return nil
 }
diff --git a/oracle/oracles/mock.go b/oracle/oracles/mock.go
index 757ce31..5ed8d75 100644
--- a/oracle/oracles/mock.go
+++ b/oracle/oracles/mock.go
@@ -36,7 +36,6 @@ package oracles
 
 import (
 	"context"
-	"math"
 	"sync"
 	"time"
 
@@ -139,23 +138,6 @@ func (o *MockOracle) SetLowResolutionTimestampUpdateInterval(time.Duration) erro
 }
 
 func (o *MockOracle) ValidateReadTS(ctx context.Context, readTS uint64, isStaleRead bool, opt *oracle.Option) error {
-	if readTS == math.MaxUint64 {
-		if isStaleRead {
-			return oracle.ErrLatestStaleRead{}
-		}
-		return nil
-	}
-
-	currentTS, err := o.GetTimestamp(ctx, opt)
-	if err != nil {
-		return errors.Errorf("fail to validate read timestamp: %v", err)
-	}
-	if currentTS < readTS {
-		return oracle.ErrFutureTSRead{
-			ReadTS:    readTS,
-			CurrentTS: currentTS,
-		}
-	}
 	return nil
 }
 
diff --git a/oracle/oracles/pd.go b/oracle/oracles/pd.go
index c2171c0..98813ad 100644
--- a/oracle/oracles/pd.go
+++ b/oracle/oracles/pd.go
@@ -662,7 +662,22 @@ func (o *pdOracle) getCurrentTSForValidation(ctx context.Context, opt *oracle.Op
 	}
 }
 
+// ValidateReadTSForTidbSnapshot is a flag in context, indicating whether the read ts is for tidb_snapshot.
+// This is a special approach for release branches to minimize code changes to reduce risks.
+type ValidateReadTSForTidbSnapshot struct{}
+
 func (o *pdOracle) ValidateReadTS(ctx context.Context, readTS uint64, isStaleRead bool, opt *oracle.Option) error {
+	// For a mistake we've seen
+	if readTS >= math.MaxInt64 && readTS < math.MaxUint64 {
+		return errors.Errorf("MaxInt64 <= readTS < MaxUint64, readTS=%v", readTS)
+	}
+
+	// only check stale reads and reads using `tidb_snapshot`
+	forTidbSnapshot := ctx.Value(ValidateReadTSForTidbSnapshot{}) != nil
+	if !forTidbSnapshot && !isStaleRead {
+		return nil
+	}
+
 	if readTS == math.MaxUint64 {
 		if isStaleRead {
 			return oracle.ErrLatestStaleRead{}
diff --git a/util/execdetails.go b/util/execdetails.go
index 5977ee9..8838cfe 100644
--- a/util/execdetails.go
+++ b/util/execdetails.go
@@ -544,6 +544,14 @@ type WriteDetail struct {
 	ApplyWriteWalDuration time.Duration
 	// ApplyWriteMemtableNanos is the time spent on writing to the memtable of the KV RocksDB.
 	ApplyWriteMemtableDuration time.Duration
+	// SchedulerLatchWaitDuration is the time spent on waiting for acquiring latch in the scheduler layer.
+	SchedulerLatchWaitDuration time.Duration
+	// SchedulerProcessDuration is the time spent on processing for the write command in the scheduler layer.
+	SchedulerProcessDuration time.Duration
+	// SchedulerThrottleDuration is the time spent on waiting due to throttled in the scheduler layer.
+	SchedulerThrottleDuration time.Duration
+	// SchedulerPessimisticLockWaitDuration is the time spent on waiting for pessimistic locks in the scheduler layer.
+	SchedulerPessimisticLockWaitDuration time.Duration
 }
 
 // MergeFromWriteDetailPb merges WriteDetail protobuf into the current WriteDetail
@@ -562,6 +570,10 @@ func (wd *WriteDetail) MergeFromWriteDetailPb(pb *kvrpcpb.WriteDetail) {
 		wd.ApplyWriteLeaderWaitDuration += time.Duration(pb.ApplyWriteLeaderWaitNanos) * time.Nanosecond
 		wd.ApplyWriteWalDuration += time.Duration(pb.ApplyWriteWalNanos) * time.Nanosecond
 		wd.ApplyWriteMemtableDuration += time.Duration(pb.ApplyWriteMemtableNanos) * time.Nanosecond
+		wd.SchedulerLatchWaitDuration += time.Duration(pb.LatchWaitNanos) * time.Nanosecond
+		wd.SchedulerProcessDuration += time.Duration(pb.ProcessNanos) * time.Nanosecond
+		wd.SchedulerThrottleDuration += time.Duration(pb.ThrottleNanos) * time.Nanosecond
+		wd.SchedulerPessimisticLockWaitDuration += time.Duration(pb.PessimisticLockWaitNanos) * time.Nanosecond
 	}
 }
 
@@ -580,6 +592,10 @@ func (wd *WriteDetail) Merge(writeDetail *WriteDetail) {
 	atomic.AddInt64((*int64)(&wd.ApplyWriteLeaderWaitDuration), int64(writeDetail.ApplyWriteLeaderWaitDuration))
 	atomic.AddInt64((*int64)(&wd.ApplyWriteWalDuration), int64(writeDetail.ApplyWriteWalDuration))
 	atomic.AddInt64((*int64)(&wd.ApplyWriteMemtableDuration), int64(writeDetail.ApplyWriteMemtableDuration))
+	atomic.AddInt64((*int64)(&wd.SchedulerLatchWaitDuration), int64(writeDetail.SchedulerLatchWaitDuration))
+	atomic.AddInt64((*int64)(&wd.SchedulerProcessDuration), int64(writeDetail.SchedulerProcessDuration))
+	atomic.AddInt64((*int64)(&wd.SchedulerThrottleDuration), int64(writeDetail.SchedulerThrottleDuration))
+	atomic.AddInt64((*int64)(&wd.SchedulerPessimisticLockWaitDuration), int64(writeDetail.SchedulerPessimisticLockWaitDuration))
 }
 
 var zeroWriteDetail = WriteDetail{}
@@ -616,6 +632,20 @@ func (wd *WriteDetail) String() string {
 	buf.WriteString(FormatDuration(wd.ApplyWriteWalDuration))
 	buf.WriteString(", write_memtable: ")
 	buf.WriteString(FormatDuration(wd.ApplyWriteMemtableDuration))
+	buf.WriteString("}, scheduler: {process: ")
+	buf.WriteString(FormatDuration(wd.SchedulerProcessDuration))
+	if wd.SchedulerLatchWaitDuration > 0 {
+		buf.WriteString(", latch_wait: ")
+		buf.WriteString(FormatDuration(wd.SchedulerLatchWaitDuration))
+	}
+	if wd.SchedulerPessimisticLockWaitDuration > 0 {
+		buf.WriteString(", pessimistic_lock_wait: ")
+		buf.WriteString(FormatDuration(wd.SchedulerPessimisticLockWaitDuration))
+	}
+	if wd.SchedulerThrottleDuration > 0 {
+		buf.WriteString(", throttle: ")
+		buf.WriteString(FormatDuration(wd.SchedulerThrottleDuration))
+	}
 	buf.WriteString("}}")
 	return buf.String()
 }


Test patch:
diff --git a/internal/locate/region_request_test.go b/internal/locate/region_request_test.go
index 0719c2c..46e067e 100644
--- a/internal/locate/region_request_test.go
+++ b/internal/locate/region_request_test.go
@@ -962,7 +962,8 @@ func (s *testRegionRequestToSingleStoreSuite) TestRegionRequestValidateReadTS()
 	testImpl(getTS, true, nil)
 	testImpl(func() uint64 { return addTS(getTS(), -time.Minute) }, false, nil)
 	testImpl(func() uint64 { return addTS(getTS(), -time.Minute) }, true, nil)
-	testImpl(func() uint64 { return addTS(getTS(), +time.Minute) }, false, oracle.ErrFutureTSRead{})
+	// check is skipped for normal read
+	testImpl(func() uint64 { return addTS(getTS(), +time.Minute) }, false, nil)
 	testImpl(func() uint64 { return addTS(getTS(), +time.Minute) }, true, oracle.ErrFutureTSRead{})
 	testImpl(func() uint64 { return math.MaxUint64 }, false, nil)
 	testImpl(func() uint64 { return math.MaxUint64 }, true, oracle.ErrLatestStaleRead{})
diff --git a/oracle/oracles/pd_test.go b/oracle/oracles/pd_test.go
index 2114710..b4c4d62 100644
--- a/oracle/oracles/pd_test.go
+++ b/oracle/oracles/pd_test.go
@@ -403,7 +403,6 @@ func TestValidateReadTS(t *testing.T) {
 	}
 
 	testImpl(true)
-	testImpl(false)
 }
 
 type MockPDClientWithPause struct {
@@ -572,8 +571,9 @@ func TestValidateReadTSForNormalReadDoNotAffectUpdateInterval(t *testing.T) {
 	mustNoNotify()
 
 	// It loads `ts + 3` and `ts + 4` from the mock PD, and the check cannot pass.
+	// Updated: 2025-03-12, the non-stale read check is temporarily skipped.
 	err = o.ValidateReadTS(ctx, ts+5, false, opt)
-	assert.Error(t, err)
+	assert.NoError(t, err)
 	mustNoNotify()
 
 	// Do the check again. It loads `ts + 5` from the mock PD, and the check passes.
@@ -618,6 +618,9 @@ func TestSetLastTSAlwaysPushTS(t *testing.T) {
 }
 
 func TestValidateReadTSFromDifferentSource(t *testing.T) {
+	// Updated: 2025-03-12, the non-stale read check is temporarily skipped.
+	t.Skip()
+
 	// If a ts is fetched from a different client to the same cluster, the ts might not be cached by the low resolution
 	// ts. In this case, the validation should not be false positive.
 	util.EnableFailpoints()