logpuller: send CDC scan priority

go.mod

@@ -52,7 +52,7 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.26
 	github.com/pingcap/errors v0.11.5-0.20260508054701-306e305bcf41
 	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86
-	github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278
+	github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c
 	github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9
 	github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5
 	github.com/pingcap/tidb v1.1.0-beta.0.20260604031706-f9faeaf4828f

go.sum

@@ -750,8 +750,8 @@ github.com/pingcap/fn v1.0.0/go.mod h1:u9WZ1ZiOD1RpNhcI42RucFh/lBuzTu6rw88a+oF2Z
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989 h1:surzm05a8C9dN8dIUmo4Be2+pMRb6f55i+UIYrluu2E=
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw=
 github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w=
-github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278 h1:VV03wSSG2XhOwhF79lvT88Z83lwzMT+aZHXrcCIN9B0=
-github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE=
+github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c h1:jeuQge1aBQp9VRboYbh+k7CCMlLI+8WT1Zustk5jpHU=
+github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE=
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM=
 github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=

logservice/eventstore/event_store.go

@@ -460,7 +460,7 @@ func (e *eventStore) Close(_ context.Context) error {
 }
 
 func (e *eventStore) RegisterDispatcher(
-	_ common.ChangeFeedID,
+	changefeedID common.ChangeFeedID,
 	dispatcherID common.DispatcherID,
 	dispatcherSpan *heartbeatpb.TableSpan,
 	startTs uint64,
@@ -478,12 +478,14 @@ func (e *eventStore) RegisterDispatcher(
 	metrics.EventStoreRegisterDispatcherStartTsLagHist.Observe(lag.Seconds())
 	if lag >= 10*time.Second {
 		log.Warn("register dispatcher with large startTs lag",
+			zap.Stringer("changefeedID", changefeedID),
 			zap.Stringer("dispatcherID", dispatcherID),
 			zap.String("span", common.FormatTableSpan(dispatcherSpan)),
 			zap.Uint64("startTs", startTs),
 			zap.Duration("lag", lag))
 	} else {
 		log.Info("register dispatcher",
+			zap.Stringer("changefeedID", changefeedID),
 			zap.Stringer("dispatcherID", dispatcherID),
 			zap.String("span", common.FormatTableSpan(dispatcherSpan)),
 			zap.Uint64("startTs", startTs))
@@ -493,6 +495,7 @@ func (e *eventStore) RegisterDispatcher(
 	defer func() {
 		if success {
 			log.Info("register dispatcher success",
+				zap.Stringer("changefeedID", changefeedID),
 				zap.Stringer("dispatcherID", dispatcherID),
 				zap.String("span", common.FormatTableSpan(dispatcherSpan)),
 				zap.Uint64("startTs", startTs),
@@ -501,6 +504,7 @@ func (e *eventStore) RegisterDispatcher(
 				zap.Duration("duration", time.Since(start)))
 		} else {
 			log.Info("register dispatcher failed",
+				zap.Stringer("changefeedID", changefeedID),
 				zap.Stringer("dispatcherID", dispatcherID),
 				zap.String("span", common.FormatTableSpan(dispatcherSpan)),
 				zap.Uint64("startTs", startTs),
@@ -549,6 +553,7 @@ func (e *eventStore) RegisterDispatcher(
 						e.addSubscriberToSubStat(subStat, dispatcherID, &Subscriber{notifyFunc: wrappedNotifier})
 						e.dispatcherMeta.Unlock()
 						log.Info("reuse existing subscription with exact span match",
+							zap.Stringer("changefeedID", changefeedID),
 							zap.Stringer("dispatcherID", dispatcherID),
 							zap.String("dispatcherSpan", common.FormatTableSpan(dispatcherSpan)),
 							zap.Uint64("startTs", startTs),
@@ -575,6 +580,7 @@ func (e *eventStore) RegisterDispatcher(
 			e.dispatcherMeta.dispatcherStats[dispatcherID] = stat
 			e.addSubscriberToSubStat(bestMatch, dispatcherID, &Subscriber{notifyFunc: wrappedNotifier})
 			log.Info("reuse existing subscription with smallest containing span",
+				zap.Stringer("changefeedID", changefeedID),
 				zap.Stringer("dispatcherID", dispatcherID),
 				zap.String("dispatcherSpan", common.FormatTableSpan(dispatcherSpan)),
 				zap.Uint64("startTs", startTs),
@@ -678,8 +684,9 @@ func (e *eventStore) RegisterDispatcher(
 	serverConfig := config.GetGlobalServerConfig()
 	resolvedTsAdvanceInterval := int64(serverConfig.KVClient.AdvanceIntervalInMs)
 	// Note: don't hold any lock when call Subscribe
-	e.subClient.Subscribe(subStat.subID, *dispatcherSpan, startTs, consumeKVEvents, advanceResolvedTs, resolvedTsAdvanceInterval, bdrMode)
+	e.subClient.Subscribe(changefeedID.String(), subStat.subID, *dispatcherSpan, startTs, consumeKVEvents, advanceResolvedTs, resolvedTsAdvanceInterval, bdrMode)
 	log.Info("new subscription created",
+		zap.Stringer("changefeedID", changefeedID),
 		zap.Stringer("dispatcherID", dispatcherID),
 		zap.Uint64("startTs", startTs),
 		zap.Uint64("subscriptionID", uint64(subStat.subID)),

logservice/eventstore/event_store_test.go

@@ -120,6 +120,7 @@ func (s *mockSubscriptionClient) AllocSubscriptionID() logpuller.SubscriptionID
 }
 
 func (s *mockSubscriptionClient) Subscribe(
+	changefeedID string,
 	subID logpuller.SubscriptionID,
 	span heartbeatpb.TableSpan,
 	startTs uint64,

logservice/logpuller/priority_task.go

@@ -14,9 +14,9 @@
 package logpuller
 
 import (
-	"fmt"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/tikv/client-go/v2/oracle"
 )
 
@@ -39,7 +39,36 @@ const (
 )
 
 func (t TaskType) String() string {
-	return fmt.Sprintf("%d", t)
+	switch t {
+	case TaskHighPrior:
+		return "high"
+	case TaskLowPrior:
+		return "low"
+	default:
+		return "unknown"
+	}
+}
+
+func (t TaskType) scanPriority() cdcpb.ScanPriority {
+	switch t {
+	case TaskHighPrior:
+		return cdcpb.ScanPriority_SCAN_PRIORITY_HIGH
+	case TaskLowPrior:
+		return cdcpb.ScanPriority_SCAN_PRIORITY_LOW
+	default:
+		return cdcpb.ScanPriority_SCAN_PRIORITY_LOW
+	}
+}
+
+func taskTypeFromScanPriority(priority cdcpb.ScanPriority) TaskType {
+	if priority == cdcpb.ScanPriority_SCAN_PRIORITY_HIGH {
+		return TaskHighPrior
+	}
+	return TaskLowPrior
+}
+
+func normalizeScanPriority(priority cdcpb.ScanPriority) cdcpb.ScanPriority {
+	return taskTypeFromScanPriority(priority).scanPriority()
 }
 
 // PriorityTask is the interface for priority-based tasks

logservice/logpuller/priority_task_test.go

@@ -18,13 +18,23 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/utils/priorityqueue"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 )
 
+func TestTaskTypeScanPriorityMapping(t *testing.T) {
+	require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, TaskHighPrior.scanPriority())
+	require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_LOW, TaskLowPrior.scanPriority())
+	require.Equal(t, TaskHighPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_HIGH))
+	require.Equal(t, TaskLowPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_LOW))
+	require.Equal(t, TaskLowPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN))
+	require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_LOW, normalizeScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN))
+}
+
 // TestPriorityCalculationLogic tests the priority calculation logic in isolation
 func TestPriorityCalculationLogic(t *testing.T) {
 	currentTime := time.Now()

logservice/logpuller/region_event_handler.go

@@ -142,6 +142,9 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent)
 			log.Panic("should not reach", zap.Any("event", event), zap.Any("events", events))
 		}
 	}
+	if newResolvedTs > 0 && h.subClient != nil {
+		h.subClient.maybeEnableRealtimeScanPriority(span, newResolvedTs)
+	}
 	tryAdvanceResolvedTs := func() {
 		if newResolvedTs != 0 {
 			span.advanceResolvedTs(newResolvedTs)

logservice/logpuller/region_req_cache_test.go

@@ -33,9 +33,10 @@ func createTestRegionInfo(subID SubscriptionID, regionID uint64) regionInfo {
 	}
 
 	subscribedSpan := &subscribedSpan{
-		subID:   subID,
-		startTs: 100,
-		span:    span,
+		subID:        subID,
+		changefeedID: "test/test-changefeed",
+		startTs:      100,
+		span:         span,
 	}
 
 	return newRegionInfo(verID, span, nil, subscribedSpan, false)

logservice/logpuller/region_request_worker.go

@@ -373,6 +373,7 @@ func (s *regionRequestWorker) processRegionSendTask(
 		subID := region.subscribedSpan.subID
 		log.Debug("region request worker gets a singleRegionInfo",
 			zap.Uint64("workerID", s.workerID),
+			zap.String("changefeedID", region.subscribedSpan.changefeedID),
 			zap.Uint64("subscriptionID", uint64(subID)),
 			zap.Uint64("regionID", region.verID.GetID()),
 			zap.String("addr", s.store.storeAddr),
@@ -448,6 +449,7 @@ func (s *regionRequestWorker) createRegionRequest(region regionInfo) *cdcpb.Chan
 		EndKey:       region.span.EndKey,
 		ExtraOp:      kvrpcpb.ExtraOp_ReadOldValue,
 		FilterLoop:   region.filterLoop,
+		ScanPriority: normalizeScanPriority(region.scanPriority),
 	}
 }
 

logservice/logpuller/region_request_worker_test.go

@@ -57,6 +57,42 @@ func prepareRegionForSendTest(region regionInfo) regionInfo {
 	return region
 }
 
+func TestCreateRegionRequestScanPriority(t *testing.T) {
+	worker := &regionRequestWorker{
+		client: &subscriptionClient{clusterID: 1},
+	}
+
+	for _, tc := range []struct {
+		name     string
+		priority cdcpb.ScanPriority
+		expected cdcpb.ScanPriority
+	}{
+		{
+			name:     "high",
+			priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+			expected: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+		},
+		{
+			name:     "low",
+			priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+			expected: cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+		},
+		{
+			name:     "unknown defaults to low",
+			priority: cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN,
+			expected: cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			region := prepareRegionForSendTest(createTestRegionInfo(1, 1))
+			region.scanPriority = tc.priority
+
+			req := worker.createRegionRequest(region)
+			require.Equal(t, tc.expected, req.GetScanPriority())
+		})
+	}
+}
+
 func TestRegionStatesOperation(t *testing.T) {
 	worker := &regionRequestWorker{}
 	worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates)

logservice/logpuller/region_state.go

@@ -16,6 +16,7 @@ package logpuller
 import (
 	"sync"
 
+	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/logservice/logpuller/regionlock"
 	"github.com/tikv/client-go/v2/tikv"
@@ -46,6 +47,9 @@ type regionInfo struct {
 	// Whether to filter out the value write by cdc itself.
 	// It should be `true` in BDR mode
 	filterLoop bool
+	// scanPriority is sent to TiKV/CSE so remote incremental scan admission can
+	// preserve TiCDC's business priority across retries.
+	scanPriority cdcpb.ScanPriority
 }
 
 func (s *regionInfo) isStopped() bool {
@@ -66,6 +70,7 @@ func newRegionInfo(
 		rpcCtx:         rpcCtx,
 		subscribedSpan: subscribedSpan,
 		filterLoop:     filterLoop,
+		scanPriority:   TaskLowPrior.scanPriority(),
 	}
 }