[DNM] hack pd client to support basic keyspace

logservice/logpuller/region_event_handler.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/spanz"
 	"github.com/pingcap/ticdc/utils/dynstream"
 	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
@@ -235,10 +236,19 @@ func handleEventEntries(span *subscribedSpan, state *regionFeedState, entries *c
 				log.Fatal("The CommitTs must be greater than the resolvedTs",
 					zap.String("EventType", "COMMITTED"),
 					zap.Uint64("CommitTs", entry.CommitTs),
-					zap.Uint64("resolvedTs", resolvedTs),
-					zap.Uint64("regionID", regionID))
+					zap.Uint64("resolvedTs", resolvedTs))
 			}
 			span.kvEventsCache = append(span.kvEventsCache, assembleRowEvent(regionID, entry))
+			key := newMatchKey(entry)
+			if state.matcher.unmatchedValue != nil {
+				if _, exists := state.matcher.unmatchedValue[key]; exists {
+					log.Warn("receive a COMMITTED entry while there is still unmatched prewrite on the same key",
+						zap.Uint64("regionID", regionID),
+						zap.String("key", spanz.HexKey(entry.GetKey())),
+						zap.Uint64("startTs", entry.GetStartTs()),
+						zap.Uint64("commitTs", entry.GetCommitTs()))
+				}
+			}
 		case cdcpb.Event_PREWRITE:
 			state.matcher.putPrewriteRow(entry)
 		case cdcpb.Event_COMMIT:

logservice/logpuller/region_request_worker.go

@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/config"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/security"
+	"github.com/pingcap/ticdc/pkg/spanz"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/pingcap/ticdc/pkg/version"
 	"go.uber.org/zap"
@@ -406,12 +407,19 @@ func (s *regionRequestWorker) processRegionSendTask(
 }
 
 func (s *regionRequestWorker) createRegionRequest(region regionInfo) *cdcpb.ChangeDataRequest {
+	resolvedTs := region.resolvedTs()
+	log.Info("region worker start to send request to store",
+		zap.Uint64("regionID", region.verID.GetID()),
+		zap.Uint64("startTs", resolvedTs),
+		zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)),
+		zap.String("startKey", spanz.HexKey(region.span.StartKey)),
+		zap.String("endKey", spanz.HexKey(region.span.EndKey)))
 	return &cdcpb.ChangeDataRequest{
 		Header:       &cdcpb.Header{ClusterId: s.client.clusterID, TicdcVersion: version.ReleaseSemver()},
 		RegionId:     region.verID.GetID(),
 		RequestId:    uint64(region.subscribedSpan.subID),
 		RegionEpoch:  region.rpcCtx.Meta.RegionEpoch,
-		CheckpointTs: region.resolvedTs(),
+		CheckpointTs: resolvedTs,
 		StartKey:     region.span.StartKey,
 		EndKey:       region.span.EndKey,
 		ExtraOp:      kvrpcpb.ExtraOp_ReadOldValue,

logservice/logpuller/txn_matcher.go

@@ -19,6 +19,7 @@ import (
 	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/spanz"
 	"go.uber.org/zap"
 )
 
@@ -127,8 +128,11 @@ func (m *matcher) matchCachedRow(initialized bool) []*cdcpb.Event_Row {
 			// prewrite log before initialized, a committed log  with
 			// the same key and start-ts must have been received.
 			log.Info("ignore commit event without prewrite",
-				zap.Binary("key", cacheEntry.GetKey()),
-				zap.Uint64("startTs", cacheEntry.GetStartTs()))
+				zap.String("key", spanz.HexKey(cacheEntry.GetKey())),
+				zap.Uint64("startTs", cacheEntry.GetStartTs()),
+				zap.Uint64("commitTs", cacheEntry.GetCommitTs()),
+				zap.Any("opType", cacheEntry.GetOpType()),
+				zap.Any("logType", cacheEntry.GetType()))
 			continue
 		}
 		cachedCommit[top] = cacheEntry

logservice/schemastore/ddl_job_fetcher.go

@@ -23,8 +23,11 @@ import (
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/logservice/logpuller"
 	"github.com/pingcap/ticdc/pkg/common"
+	appctx "github.com/pingcap/ticdc/pkg/common/context"
 	"github.com/pingcap/ticdc/pkg/common/event"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/pdutil"
+	"github.com/pingcap/ticdc/pkg/spanz"
 	"github.com/pingcap/ticdc/utils/heap"
 	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/kv"
@@ -124,7 +127,7 @@ func (p *ddlJobFetcher) input(kvs []common.RawKVEntry, _ func()) bool {
 	for _, kv := range kvs {
 		job, err := p.unmarshalDDL(&kv)
 		if err != nil {
-			log.Fatal("unmarshal ddl failed", zap.Any("kv", kv), zap.Error(err))
+			log.Fatal("unmarshal ddl failed", zap.String("key", spanz.HexKey(kv.Key)), zap.Any("kv", kv), zap.Error(err))
 		}
 
 		if job == nil {
@@ -258,14 +261,29 @@ const (
 )
 
 func getAllDDLSpan() []heartbeatpb.TableSpan {
+	pdClient := appctx.GetService[pdutil.PDAPIClient](appctx.PDAPIClient)
+	keyspaceID, err := pdClient.GetKeyspaceID(context.Background(), "SYSTEM")
+	if err != nil {
+		log.Panic("get keyspace id from pd client failed", zap.Error(err))
+	}
+
 	spans := make([]heartbeatpb.TableSpan, 0, 2)
-	start, end := common.GetTableRange(JobTableID)
+	start, end, err := common.GetKeyspaceTableRange(keyspaceID, JobTableID)
+	if err != nil {
+		log.Panic("get keyspace table range failed",
+			zap.Uint32("keyspaceID", keyspaceID), zap.Error(err))
+	}
 	spans = append(spans, heartbeatpb.TableSpan{
 		TableID:  JobTableID,
 		StartKey: common.ToComparableKey(start),
 		EndKey:   common.ToComparableKey(end),
 	})
-	start, end = common.GetTableRange(JobHistoryID)
+
+	start, end, err = common.GetKeyspaceTableRange(keyspaceID, JobHistoryID)
+	if err != nil {
+		log.Panic("get keyspace table range failed",
+			zap.Uint32("keyspaceID", keyspaceID), zap.Error(err))
+	}
 	spans = append(spans, heartbeatpb.TableSpan{
 		TableID:  JobHistoryID,
 		StartKey: common.ToComparableKey(start),

maintainer/maintainer_controller_bootstrap.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/filter"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/ticdc/utils"
 	"go.uber.org/zap"
 )
@@ -209,7 +210,16 @@ func (c *Controller) processTableSpans(
 	// Add new table if not working
 	if isTableWorking {
 		// Handle existing table spans
-		span := common.TableIDToComparableSpan(table.TableID)
+		pdClient := appcontext.GetService[pdutil.PDAPIClient](appcontext.PDAPIClient)
+		keyspaceID, err := pdClient.GetKeyspaceID(context.Background(), "SYSTEM")
+		if err != nil {
+			log.Panic("get codec from pd client failed", zap.Error(err))
+		}
+		span, err := common.TableIDToComparableSpanWithKeyspace(keyspaceID, table.TableID)
+		if err != nil {
+			log.Panic("tableIDToComparableSpanWithKeyspace failed",
+				zap.Uint32("keyspaceID", keyspaceID), zap.Int64("tableID", table.TableID), zap.Error(err))
+		}
 		tableSpan := &heartbeatpb.TableSpan{
 			TableID:  table.TableID,
 			StartKey: span.StartKey,

maintainer/maintainer_controller_helper.go

@@ -25,8 +25,10 @@ import (
 	"github.com/pingcap/ticdc/maintainer/span"
 	"github.com/pingcap/ticdc/maintainer/split"
 	"github.com/pingcap/ticdc/pkg/common"
+	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	pkgoperator "github.com/pingcap/ticdc/pkg/scheduler/operator"
 	"go.uber.org/zap"
 )
@@ -165,7 +167,16 @@ func (c *Controller) splitTableByRegionCount(tableID int64, mode int64) error {
 		return nil
 	}
 
-	span := common.TableIDToComparableSpan(tableID)
+	pdClient := appcontext.GetService[pdutil.PDAPIClient](appcontext.PDAPIClient)
+	keyspaceID, err := pdClient.GetKeyspaceID(context.Background(), "SYSTEM")
+	if err != nil {
+		log.Panic("get codec from pd client failed", zap.Error(err))
+	}
+	span, err := common.TableIDToComparableSpanWithKeyspace(keyspaceID, tableID)
+	if err != nil {
+		log.Panic("tableIDToComparableSpanWithKeyspace failed",
+			zap.Uint32("keyspaceID", keyspaceID), zap.Int64("tableID", tableID), zap.Error(err))
+	}
 	wholeSpan := &heartbeatpb.TableSpan{
 		TableID:  span.TableID,
 		StartKey: span.StartKey,

maintainer/range_checker/table_span_range_checker.go

@@ -15,12 +15,17 @@ package range_checker
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"strings"
 	"sync/atomic"
 
 	"github.com/google/btree"
+	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
+	appcontext "github.com/pingcap/ticdc/pkg/common/context"
+	"github.com/pingcap/ticdc/pkg/pdutil"
+	"go.uber.org/zap"
 )
 
 // TableSpanRangeChecker is used to check if all ranges cover the start and end byte slices.
@@ -35,8 +40,17 @@ func NewTableSpanRangeChecker(tables []int64) *TableSpanRangeChecker {
 		tableSpans: make(map[int64]*SpanCoverageChecker),
 		covered:    atomic.Bool{},
 	}
+	pdClient := appcontext.GetService[pdutil.PDAPIClient](appcontext.PDAPIClient)
+	keyspaceID, err := pdClient.GetKeyspaceID(context.Background(), "SYSTEM")
+	if err != nil {
+		log.Panic("get codec from pd client failed", zap.Error(err))
+	}
 	for _, table := range tables {
-		span := common.TableIDToComparableSpan(table)
+		span, err := common.TableIDToComparableSpanWithKeyspace(keyspaceID, table)
+		if err != nil {
+			log.Panic("tableIDToComparableSpanWithKeyspace failed",
+				zap.Uint32("keyspaceID", keyspaceID), zap.Error(err))
+		}
 		sc.tableSpans[table] = NewTableSpanCoverageChecker(span.StartKey, span.EndKey)
 	}
 	return sc

maintainer/replica/replication_span.go

@@ -15,14 +15,17 @@ package replica
 
 import (
 	"bytes"
+	"context"
 	"encoding/hex"
 	"sync"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/pkg/common"
+	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	"github.com/pingcap/ticdc/pkg/messaging"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/pingcap/ticdc/pkg/scheduler/replica"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
@@ -127,8 +130,18 @@ func (r *SpanReplication) initStatus(status *heartbeatpb.TableSpanStatus) {
 func (r *SpanReplication) initGroupID() {
 	r.groupID = replica.DefaultGroupID
 	span := heartbeatpb.TableSpan{TableID: r.Span.TableID, StartKey: r.Span.StartKey, EndKey: r.Span.EndKey}
+	pdClient := appcontext.GetService[pdutil.PDAPIClient](appcontext.PDAPIClient)
+	keyspaceID, err := pdClient.GetKeyspaceID(context.Background(), "SYSTEM")
+	if err != nil {
+		log.Panic("get codec from pd client failed", zap.Error(err))
+	}
 	// check if the table is split
-	totalSpan := common.TableIDToComparableSpan(span.TableID)
+	totalSpan, err := common.TableIDToComparableSpanWithKeyspace(keyspaceID, span.TableID)
+	if err != nil {
+		log.Panic("tableIDToComparableSpanWithKeyspace failed",
+			zap.Uint32("keyspaceID", keyspaceID), zap.Int64("tableID", span.TableID), zap.Error(err))
+	}
+
 	if !common.IsSubSpan(span, totalSpan) {
 		log.Warn("invalid span range", zap.String("changefeedID", r.ChangefeedID.Name()),
 			zap.String("id", r.ID.String()), zap.Int64("tableID", span.TableID),

maintainer/span/span_controller.go

@@ -27,6 +27,7 @@ import (
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/pdutil"
 	pkgreplica "github.com/pingcap/ticdc/pkg/scheduler/replica"
 	"github.com/pingcap/ticdc/server/watcher"
 	"github.com/pingcap/ticdc/utils"
@@ -142,7 +143,16 @@ func (c *Controller) AddNewTable(table commonEvent.Table, startTs uint64) {
 			zap.Int64("table", table.TableID))
 		return
 	}
-	span := common.TableIDToComparableSpan(table.TableID)
+	pdClient := appcontext.GetService[pdutil.PDAPIClient](appcontext.PDAPIClient)
+	keyspaceID, err := pdClient.GetKeyspaceID(context.Background(), "SYSTEM")
+	if err != nil {
+		log.Panic("get keyspaceID from pd client failed", zap.Error(err))
+	}
+	span, err := common.TableIDToComparableSpanWithKeyspace(keyspaceID, table.TableID)
+	if err != nil {
+		log.Panic("tableIDToComparableSpanWithKeyspace failed",
+			zap.Uint32("keyspaceID", keyspaceID), zap.Int64("tableID", table.TableID), zap.Error(err))
+	}
 	tableSpan := &heartbeatpb.TableSpan{
 		TableID:  table.TableID,
 		StartKey: span.StartKey,

maintainer/testutil/test_util.go

@@ -133,6 +133,10 @@ func (m *MockPDAPIClient) ScanRegions(ctx context.Context, span heartbeatpb.Tabl
 	return m.scnaRegionsResult[fmt.Sprintf("%s-%s", span.StartKey, span.EndKey)], nil
 }
 
+func (m *MockPDAPIClient) GetKeyspaceID(ctx context.Context, keyspace string) (uint32, error) {
+	return 0, nil
+}
+
 func (m *MockPDAPIClient) Close() {
 	// Mock implementation - do nothing
 }

pkg/common/event/mounter.go

@@ -157,6 +157,24 @@ func IsLegacyFormatJob(rawKV *common.RawKVEntry) bool {
 	return bytes.HasPrefix(rawKV.Key, metaPrefix)
 }
 
+const (
+	keyspacePrefixLen       = 4
+	apiV2TxnModePrefix byte = 'x'
+)
+
+// RemoveKeyspacePrefix is used to remove keyspace prefix from the key.
+func RemoveKeyspacePrefix(key []byte) []byte {
+	if len(key) <= keyspacePrefixLen {
+		return key
+	}
+
+	if key[0] != apiV2TxnModePrefix {
+		log.Warn("the first byte of key is not 'x', it may not be in api v2 txn mode", zap.Any("byte", key[0]))
+		return key
+	}
+	return key[keyspacePrefixLen:]
+}
+
 // ParseDDLJob parses the job from the raw KV entry.
 func ParseDDLJob(rawKV *common.RawKVEntry, ddlTableInfo *DDLTableInfo) (*model.Job, error) {
 	var v []byte
@@ -172,6 +190,7 @@ func ParseDDLJob(rawKV *common.RawKVEntry, ddlTableInfo *DDLTableInfo) (*model.J
 		return job, err
 	}
 
+	rawKV.Key = RemoveKeyspacePrefix(rawKV.Key)
 	recordID, err := tablecodec.DecodeRowKey(rawKV.Key)
 	if err != nil {
 		return nil, errors.Trace(err)

pkg/common/event/mounter_test.go

@@ -14,11 +14,13 @@
 package event
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/integrity"
+	"github.com/pingcap/ticdc/pkg/spanz"
 	timodel "github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/types"