Merge branch 'master' into set-u

pingcap · Dec 31, 2021 · 143d5a9 · 143d5a9
2 parents f2c509d + 4608f6a
commit 143d5a9
Show file tree

Hide file tree

Showing 7 changed files with 487 additions and 170 deletions.
diff --git a/cdc/processor/pipeline/sorter.go b/cdc/processor/pipeline/sorter.go
@@ -112,7 +112,10 @@ func (n *sorterNode) StartActorNode(ctx pipeline.NodeContext, isTableActorMode b
 			startTs := ctx.ChangefeedVars().Info.StartTs
 			actorID := ctx.GlobalVars().SorterSystem.ActorID(uint64(n.tableID))
 			router := ctx.GlobalVars().SorterSystem.Router()
-			levelSorter := leveldb.NewSorter(ctx, n.tableID, startTs, router, actorID)
+			compactScheduler := ctx.GlobalVars().SorterSystem.CompactScheduler()
+			levelSorter := leveldb.NewSorter(
+				ctx, n.tableID, startTs, router, actorID, compactScheduler,
+				config.GetGlobalServerConfig().Debug.DB)
 			n.cleanID = actorID
 			n.cleanTask = levelSorter.CleanupTask()
 			n.cleanRouter = ctx.GlobalVars().SorterSystem.CleanerRouter()

diff --git a/cdc/sorter/leveldb/system/system.go b/cdc/sorter/leveldb/system/system.go
@@ -107,6 +107,11 @@ func (s *System) CleanerRouter() *actor.Router {
 	return s.cleanRouter
 }
 
+// CompactScheduler returns compaction scheduler.
+func (s *System) CompactScheduler() *lsorter.CompactScheduler {
+	return s.compactSched
+}
+
 // broadcase messages to actors in the router.
 // Caveats it may lose messages quietly.
 func (s *System) broadcast(ctx context.Context, router *actor.Router, msg message.Message) {
@@ -139,7 +144,7 @@ func (s *System) Start(ctx context.Context) error {
 	dbCount := s.cfg.Count
 	for id := 0; id < dbCount; id++ {
 		// Open db.
-		db, err := db.OpenLevelDB(ctx, id, s.dir, s.cfg)
+		db, err := db.OpenPebble(ctx, id, s.dir, s.cfg)
 		if err != nil {
 			return errors.Trace(err)
 		}