Repository: pingcap/tiflow
Pull request #4058 (2021-12-24 08:50:45)

db,sorter/leveldb(ticdc): remove snapshot
### What problem does this PR solve?

Remove useless snapshot API.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

### Release note <!-- bugfixes or new feature need a release note -->

```release-note
None
```


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4058/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4058/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

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

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
codecov-commenter: # [Codecov](https://codecov.io/gh/pingcap/tiflow/pull/4058?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#4058](https://codecov.io/gh/pingcap/tiflow/pull/4058?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (f9d052e) into [master](https://codecov.io/gh/pingcap/tiflow/commit/3873d3938927e8a801276a5ab4ae889028e5073f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (3873d39) will **decrease** coverage by `1.9330%`.
> The diff coverage is `70.9349%`.


| Flag | Coverage Δ | |
|---|---|---|
| cdc | `58.6315% <70.9349%> (+0.3949%)` | :arrow_up: |
| dm | `52.1998% <ø> (-3.8348%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

```diff
@@               Coverage Diff                @@
##             master      #4058        +/-   ##
================================================
- Coverage   57.0741%   55.1410%   -1.9331%     
================================================
  Files           478        485         +7     
  Lines         56551      59890      +3339     
================================================
+ Hits          32276      33024       +748     
- Misses        20978      23526      +2548     
- Partials       3297       3340        +43     
```

overvenus: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: f9d052e0dd1db80a206063d7b0839bdbce07855c</details>
ti-chi-bot: @overvenus: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
overvenus: /merge
ti-chi-bot: @overvenus: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
overvenus: /merge
ti-chi-bot: Merge canceled because a new commit is pushed.
overvenus: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: 1b75f5e6bb952ac8e5b51c91ef11f57455ff7b3b</details>
overvenus: /run-verify
/run-dm-integration-test
ti-chi-bot: @overvenus: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
overvenus: /run-verify
/run-kafka-integration-test
/run-integration-test
overvenus: /run-verify
/run-dm-integration-test
overvenus: /run-dm-integration-test

Patch:
diff --git a/cdc/sorter/leveldb/cleaner.go b/cdc/sorter/leveldb/cleaner.go
index 3e8eee925..18cb2ab15 100644
--- a/cdc/sorter/leveldb/cleaner.go
+++ b/cdc/sorter/leveldb/cleaner.go
@@ -103,13 +103,7 @@ TASKS:
 
 		start := encoding.EncodeTsKey(task.UID, task.TableID, 0)
 		limit := encoding.EncodeTsKey(task.UID, task.TableID+1, 0)
-		snap, err := clean.db.Snapshot()
-		if err != nil {
-			log.Panic("get snapshot failed",
-				zap.Error(err), zap.Uint64("id", uint64(clean.id)))
-			return false
-		}
-		iter := snap.Iterator(start, limit)
+		iter := clean.db.Iterator(start, limit)
 
 		// Force writes the first batch if the task is rescheduled (rate limited).
 		force := task.CleanupRatelimited
@@ -135,23 +129,13 @@ TASKS:
 						log.Panic("db error",
 							zap.Error(err), zap.Uint64("id", uint64(clean.id)))
 					}
-					err = snap.Release()
-					if err != nil {
-						log.Panic("db error",
-							zap.Error(err), zap.Uint64("id", uint64(clean.id)))
-					}
 					break TASKS
 				}
 				force = false
 			}
 		}
 		// Release iterator and snapshot in time.
-		err = iter.Release()
-		if err != nil {
-			log.Panic("db error",
-				zap.Error(err), zap.Uint64("id", uint64(clean.id)))
-		}
-		err = snap.Release()
+		err := iter.Release()
 		if err != nil {
 			log.Panic("db error",
 				zap.Error(err), zap.Uint64("id", uint64(clean.id)))
diff --git a/cdc/sorter/leveldb/leveldb.go b/cdc/sorter/leveldb/leveldb.go
index c8ca1a139..dce80ef12 100644
--- a/cdc/sorter/leveldb/leveldb.go
+++ b/cdc/sorter/leveldb/leveldb.go
@@ -14,12 +14,12 @@
 package leveldb
 
 import (
+	"container/list"
 	"context"
 	"strconv"
 	"sync"
 	"time"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/sorter/leveldb/message"
 	"github.com/pingcap/tiflow/pkg/actor"
@@ -32,6 +32,47 @@ import (
 	"golang.org/x/sync/semaphore"
 )
 
+// Queue of IterRequest
+type iterQueue struct {
+	*list.List
+	// TableID set.
+	tables map[tableKey]struct{}
+}
+
+type iterItem struct {
+	key tableKey
+	req *message.IterRequest
+}
+
+type tableKey struct {
+	UID     uint32
+	TableID uint64
+}
+
+func (q *iterQueue) push(uid uint32, tableID uint64, req *message.IterRequest) {
+	key := tableKey{UID: uid, TableID: tableID}
+	_, ok := q.tables[key]
+	if ok {
+		log.Panic("A table should not issue two concurrent iterator requests",
+			zap.Uint64("tableID", tableID),
+			zap.Uint32("UID", uid),
+			zap.Uint64("resolvedTs", req.ResolvedTs))
+	}
+	q.tables[key] = struct{}{}
+	q.List.PushBack(iterItem{req: req, key: key})
+}
+
+func (q *iterQueue) pop() (*message.IterRequest, bool) {
+	item := q.List.Front()
+	if item == nil {
+		return nil, false
+	}
+	q.List.Remove(item)
+	req := item.Value.(iterItem)
+	delete(q.tables, req.key)
+	return req.req, true
+}
+
 // DBActor is a db actor, it reads, writes and deletes key value pair in its db.
 type DBActor struct {
 	id      actor.ID
@@ -39,7 +80,8 @@ type DBActor struct {
 	wb      db.Batch
 	wbSize  int
 	wbCap   int
-	snapSem *semaphore.Weighted
+	iterSem *semaphore.Weighted
+	iterQ   iterQueue
 
 	deleteCount int
 	compact     *CompactScheduler
@@ -71,11 +113,15 @@ func NewDBActor(
 	mb := actor.NewMailbox(actor.ID(id), cfg.Concurrency)
 	wg.Add(1)
 
-	dba := &DBActor{
+	return &DBActor{
 		id:      actor.ID(id),
 		db:      db,
 		wb:      wb,
-		snapSem: iterSema,
+		iterSem: iterSema,
+		iterQ: iterQueue{
+			List:   list.New(),
+			tables: make(map[tableKey]struct{}),
+		},
 		wbSize:  wbSize,
 		wbCap:   wbCap,
 		compact: compact,
@@ -84,8 +130,7 @@ func NewDBActor(
 
 		metricWriteDuration: sorterWriteDurationHistogram.WithLabelValues(captureAddr, idTag),
 		metricWriteBytes:    sorterWriteBytesHistogram.WithLabelValues(captureAddr, idTag),
-	}
-	return dba, mb, nil
+	}, mb, nil
 }
 
 func (ldb *DBActor) close(err error) {
@@ -120,6 +165,31 @@ func (ldb *DBActor) maybeWrite(force bool) error {
 	return nil
 }
 
+// Batch acquire iterators for requests in the queue.
+func (ldb *DBActor) acquireIterators() {
+	for {
+		succeed := ldb.iterSem.TryAcquire(1)
+		if !succeed {
+			break
+		}
+		req, ok := ldb.iterQ.pop()
+		if !ok {
+			ldb.iterSem.Release(1)
+			break
+		}
+
+		iterCh := req.IterCh
+		iterRange := req.Range
+		iter := ldb.db.Iterator(iterRange[0], iterRange[1])
+		iterCh <- &message.LimitedIterator{
+			Iterator:   iter,
+			Sema:       ldb.iterSem,
+			ResolvedTs: req.ResolvedTs,
+		}
+		close(iterCh)
+	}
+}
+
 // Poll implements actor.Actor.
 // It handles tasks by writing kv, deleting kv and taking iterators.
 func (ldb *DBActor) Poll(ctx context.Context, tasks []actormsg.Message) bool {
@@ -129,11 +199,13 @@ func (ldb *DBActor) Poll(ctx context.Context, tasks []actormsg.Message) bool {
 		return false
 	default:
 	}
-	snapChs := make([]chan message.LimitedSnapshot, 0, len(tasks))
+	requireIter := false
 	for i := range tasks {
 		var task message.Task
 		msg := tasks[i]
 		switch msg.Tp {
+		case actormsg.TypeTick:
+			continue
 		case actormsg.TypeSorterTask:
 			task = msg.SorterTask
 		case actormsg.TypeStop:
@@ -142,9 +214,8 @@ func (ldb *DBActor) Poll(ctx context.Context, tasks []actormsg.Message) bool {
 		default:
 			log.Panic("unexpected message", zap.Any("message", msg))
 		}
-		events, needSnap, snapCh := task.Events, task.NeedSnap, task.SnapCh
 
-		for k, v := range events {
+		for k, v := range task.Events {
 			if len(v) != 0 {
 				ldb.wb.Put([]byte(k), v)
 			} else {
@@ -158,42 +229,18 @@ func (ldb *DBActor) Poll(ctx context.Context, tasks []actormsg.Message) bool {
 				log.Panic("db error", zap.Error(err))
 			}
 		}
-		if needSnap {
+		if task.IterReq != nil {
 			// Append to slice for later batch acquiring iterators.
-			snapChs = append(snapChs, snapCh)
-		} else {
-			// Or close channel to notify caller that that its task is done.
-			close(snapCh)
+			ldb.iterQ.push(task.UID, task.TableID, task.IterReq)
+			requireIter = true
 		}
 	}
 
 	// Force write only if there is a task requires an iterator.
-	forceWrite := len(snapChs) != 0
-	err := ldb.maybeWrite(forceWrite)
-	if err != nil {
+	if err := ldb.maybeWrite(requireIter); err != nil {
 		log.Panic("db error", zap.Error(err))
 	}
-	// Batch acquire iterators.
-	for i := range snapChs {
-		snapCh := snapChs[i]
-		err := ldb.snapSem.Acquire(ctx, 1)
-		if err != nil {
-			if errors.Cause(err) == context.Canceled ||
-				errors.Cause(err) == context.DeadlineExceeded {
-				return false
-			}
-			log.Panic("db unreachable error, acquire iter", zap.Error(err))
-		}
-		snap, err := ldb.db.Snapshot()
-		if err != nil {
-			log.Panic("db error", zap.Error(err))
-		}
-		snapCh <- message.LimitedSnapshot{
-			Snapshot: snap,
-			Sema:     ldb.snapSem,
-		}
-		close(snapCh)
-	}
+	ldb.acquireIterators()
 
 	return true
 }
diff --git a/cdc/sorter/leveldb/message/task.go b/cdc/sorter/leveldb/message/task.go
index 92595bcb4..319d85f1e 100644
--- a/cdc/sorter/leveldb/message/task.go
+++ b/cdc/sorter/leveldb/message/task.go
@@ -30,16 +30,26 @@ type Task struct {
 	// encoded key -> serde.marshal(event)
 	// If a value is empty, it deletes the key/value entry in db.
 	Events map[Key][]byte
-	// Must be buffered channel to avoid blocking.
-	SnapCh chan LimitedSnapshot `json:"-"` // Make Task JSON printable.
-	// Set NeedSnap whenever caller wants to read something from a snapshot.
-	NeedSnap bool
+	// Requests an iterator when it is not nil.
+	IterReq *IterRequest
 
 	// For clean-up table task.
 	Cleanup            bool
 	CleanupRatelimited bool
 }
 
+// IterRequest contains parameters that necessary to build an iterator.
+type IterRequest struct {
+	UID uint32
+
+	// The resolved ts at the time of issuing the request.
+	ResolvedTs uint64
+	// Range of a requested iterator.
+	Range [2][]byte
+	// Must be buffered channel to avoid blocking.
+	IterCh chan *LimitedIterator `json:"-"` // Make Task JSON printable.
+}
+
 // Key is the key that is written to db.
 type Key string
 
@@ -51,17 +61,18 @@ func (k Key) String() string {
 		uid, tableID, startTs, CRTs)
 }
 
-// LimitedSnapshot is a wrapper of db.Snapshot that has a sema to limit
-// the total number of alive snapshots.
-type LimitedSnapshot struct {
-	db.Snapshot
-	Sema *semaphore.Weighted
+// LimitedIterator is a wrapper of db.Iterator that has a sema to limit
+// the total number of alive iterator.
+type LimitedIterator struct {
+	db.Iterator
+	Sema       *semaphore.Weighted
+	ResolvedTs uint64
 }
 
 // Release resources of the snapshot.
-func (s *LimitedSnapshot) Release() error {
+func (s *LimitedIterator) Release() error {
 	s.Sema.Release(1)
-	return errors.Trace(s.Snapshot.Release())
+	return errors.Trace(s.Iterator.Release())
 }
 
 // NewCleanupTask returns a clean up task to clean up table data.
diff --git a/cdc/sorter/leveldb/metrics.go b/cdc/sorter/leveldb/metrics.go
index eb3705274..ff12d5c7d 100644
--- a/cdc/sorter/leveldb/metrics.go
+++ b/cdc/sorter/leveldb/metrics.go
@@ -42,6 +42,14 @@ var (
 		Buckets:   prometheus.ExponentialBuckets(0.004, 2.0, 20),
 	}, []string{"capture", "id"})
 
+	sorterIterReadDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "ticdc",
+		Subsystem: "sorter",
+		Name:      "db_iter_read_duration_seconds",
+		Help:      "Bucketed histogram of db sorter iterator read duration",
+		Buckets:   prometheus.ExponentialBuckets(0.004, 2.0, 20),
+	}, []string{"capture", "id", "call"})
+
 	sorterCleanupKVCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "ticdc",
 		Subsystem: "sorter",
@@ -55,5 +63,6 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(sorterWriteDurationHistogram)
 	registry.MustRegister(sorterCompactDurationHistogram)
 	registry.MustRegister(sorterWriteBytesHistogram)
+	registry.MustRegister(sorterIterReadDurationHistogram)
 	registry.MustRegister(sorterCleanupKVCounter)
 }
diff --git a/cdc/sorter/leveldb/system/system.go b/cdc/sorter/leveldb/system/system.go
index 580d671ee..dcd42331b 100644
--- a/cdc/sorter/leveldb/system/system.go
+++ b/cdc/sorter/leveldb/system/system.go
@@ -52,6 +52,7 @@ type System struct {
 	cleanRouter   *actor.Router
 	compactSystem *actor.System
 	compactRouter *actor.Router
+	compactSched  *lsorter.CompactScheduler
 	dir           string
 	cfg           *config.DBConfig
 	closedCh      chan struct{}
@@ -69,6 +70,7 @@ func NewSystem(dir string, cfg *config.DBConfig) *System {
 		WorkerNumber(cfg.Count).Build()
 	compactSystem, compactRouter := actor.NewSystemBuilder("compactor").
 		WorkerNumber(cfg.Count).Build()
+	compactSched := lsorter.NewCompactScheduler(compactRouter, cfg)
 	return &System{
 		dbSystem:      dbSystem,
 		dbRouter:      dbRouter,
@@ -76,6 +78,7 @@ func NewSystem(dir string, cfg *config.DBConfig) *System {
 		cleanRouter:   cleanRouter,
 		compactSystem: compactSystem,
 		compactRouter: compactRouter,
+		compactSched:  compactSched,
 		dir:           dir,
 		cfg:           cfg,
 		closedCh:      make(chan struct{}),
@@ -151,10 +154,9 @@ func (s *System) Start(ctx context.Context) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
-		compact := lsorter.NewCompactScheduler(s.compactRouter, s.cfg)
 		// Create and spawn db actor.
 		dbac, dbmb, err :=
-			lsorter.NewDBActor(id, db, s.cfg, compact, s.closedWg, captureAddr)
+			lsorter.NewDBActor(id, db, s.cfg, s.compactSched, s.closedWg, captureAddr)
 		if err != nil {
 			return errors.Trace(err)
 		}
@@ -164,7 +166,7 @@ func (s *System) Start(ctx context.Context) error {
 		}
 		// Create and spawn cleaner actor.
 		clac, clmb, err := lsorter.NewCleanerActor(
-			id, db, s.cleanRouter, compact, s.cfg, s.closedWg)
+			id, db, s.cleanRouter, s.compactSched, s.cfg, s.closedWg)
 		if err != nil {
 			return errors.Trace(err)
 		}
diff --git a/cdc/sorter/leveldb/table_sorter.go b/cdc/sorter/leveldb/table_sorter.go
index 332d7d29e..3dd0ace78 100644
--- a/cdc/sorter/leveldb/table_sorter.go
+++ b/cdc/sorter/leveldb/table_sorter.go
@@ -206,19 +206,15 @@ BATCH:
 	return maxCommitTs, maxResolvedTs, inputCount, nil
 }
 
-// asyncWrite writes events and delete keys asynchronously.
-// It returns a channel to notify caller when write is done,
-// if needSnap is true, caller receives a snapshot and reads all resolved
-// events, up to the maxResolvedTs.
-func (ls *Sorter) asyncWrite(
-	ctx context.Context, events []*model.PolymorphicEvent, deleteKeys []message.Key, needSnap bool,
-) (chan message.LimitedSnapshot, error) {
+// buildTask build a task for writing new events and delete outputted events.
+func (ls *Sorter) buildTask(
+	events []*model.PolymorphicEvent, deleteKeys []message.Key,
+) (message.Task, error) {
 	// Write and sort events.
 	tk := message.Task{
 		UID:     ls.uid,
 		TableID: ls.tableID,
 		Events:  make(map[message.Key][]byte),
-		SnapCh:  make(chan message.LimitedSnapshot, 1),
 	}
 	for i := range events {
 		event := events[i]
@@ -231,7 +227,7 @@ func (ls *Sorter) asyncWrite(
 		var err error
 		value, err = ls.serde.Marshal(event, value)
 		if err != nil {
-			return nil, errors.Trace(err)
+			return message.Task{}, errors.Trace(err)
 		}
 		tk.Events[message.Key(key)] = value
 	}
@@ -241,11 +237,7 @@ func (ls *Sorter) asyncWrite(
 		tk.Events[deleteKeys[i]] = []byte{}
 	}
 
-	tk.NeedSnap = needSnap
-
-	// Send write task to leveldb.
-	err := ls.router.SendB(ctx, ls.actorID, actormsg.SorterMessage(tk))
-	return tk.SnapCh, errors.Trace(err)
+	return tk, nil
 }
 
 // output nonblocking outputs an event. Caller should retry when it returns false.
@@ -468,13 +460,12 @@ func (ls *Sorter) poll(ctx context.Context, state *pollState) error {
 	// It can only acquire a snapshot when
 	// 1. No buffered resolved events, they must be sent before
 	//    sending further resolved events from snapshot.
-	needSnap := lenResolvedEvents == 0
+	needIter := lenResolvedEvents == 0
 	// 2. There are some events that can be resolved.
-	needSnap = needSnap && state.hasResolvedEvents()
+	needIter = needIter && state.hasResolvedEvents()
 
 	// Write new events and delete sent keys.
-	snapCh, err :=
-		ls.asyncWrite(ctx, newEvents, state.outputBuf.deleteKeys, needSnap)
+	task, err := ls.buildTask(newEvents, state.outputBuf.deleteKeys)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -483,40 +474,49 @@ func (ls *Sorter) poll(ctx context.Context, state *pollState) error {
 	// Try shrink buffer to release memory.
 	state.outputBuf.maybeShrink()
 
-	if !needSnap {
+	if !needIter {
 		// No new events and no resolved events.
 		if !state.hasResolvedEvents() && state.maxResolvedTs != 0 {
 			ls.outputResolvedTs(state.maxResolvedTs)
 		}
-		return nil
+		// Send write task to leveldb.
+		return ls.router.SendB(ctx, ls.actorID, actormsg.SorterMessage(task))
+	}
+
+	iterCh := make(chan *message.LimitedIterator, 1)
+	task.IterReq = &message.IterRequest{
+		UID:        ls.uid,
+		ResolvedTs: state.maxResolvedTs,
+		Range: [2][]byte{
+			encoding.EncodeTsKey(ls.uid, ls.tableID, 0),
+			encoding.EncodeTsKey(ls.uid, ls.tableID, state.maxResolvedTs+1),
+		},
+		IterCh: iterCh,
+	}
+	// Send write task to leveldb.
+	err = ls.router.SendB(ctx, ls.actorID, actormsg.SorterMessage(task))
+	if err != nil {
+		return errors.Trace(err)
 	}
 
 	// Wait for writing and deleting.
-	var snap message.LimitedSnapshot
+	var iter *message.LimitedIterator
 	var ok bool
 	select {
 	case <-ctx.Done():
 		return errors.Trace(ctx.Err())
-	case snap, ok = <-snapCh:
+	case iter, ok = <-iterCh:
 	}
-	if !ok && needSnap {
+	if !ok && needIter {
 		return cerrors.ErrUnexpectedSnapshot.GenWithStackByArgs(ls.tableID)
 	}
-	iter := snap.Iterator(
-		encoding.EncodeTsKey(ls.uid, ls.tableID, 0),
-		encoding.EncodeTsKey(ls.uid, ls.tableID, state.maxResolvedTs+1),
-	)
 	defer func() {
-		// Release iter and snap should be fast, otherwise it may block
+		// Release iter should be fast, otherwise it may block
 		// the funcation.
 		if err := iter.Release(); err != nil {
 			log.Error("iterator release error",
 				zap.Error(err), zap.Uint64("tableID", ls.tableID))
 		}
-		if err := snap.Release(); err != nil {
-			log.Error("snapshot release error",
-				zap.Error(err), zap.Uint64("tableID", ls.tableID))
-		}
 	}()
 
 	// Read and send resolved events from iterator.
diff --git a/pkg/config/db.go b/pkg/config/db.go
index b2464fc35..ca9e0b447 100644
--- a/pkg/config/db.go
+++ b/pkg/config/db.go
@@ -13,7 +13,9 @@
 
 package config
 
-import cerror "github.com/pingcap/tiflow/pkg/errors"
+import (
+	cerror "github.com/pingcap/tiflow/pkg/errors"
+)
 
 // DBConfig represents leveldb sorter config.
 type DBConfig struct {
@@ -60,6 +62,7 @@ type DBConfig struct {
 	//
 	// The default value is 1<<31 - 1.
 	WriteL0PauseTrigger int `toml:"write-l0-pause-trigger" json:"write-l0-pause-trigger"`
+
 	// CompactionL0Trigger defines number of leveldb sst file at level-0 that will
 	// trigger compaction.
 	//
@@ -72,6 +75,18 @@ type DBConfig struct {
 	// Iterator.First() takes about 27ms to 149ms in this case,
 	// see pkg/db.BenchmarkNext.
 	CompactionDeletionThreshold int `toml:"compaction-deletion-threshold" json:"compaction-deletion-threshold"`
+
+	// IteratorMaxAliveDuration the maximum iterator alive duration in ms.
+	//
+	// The default value is 10000, 10s
+	IteratorMaxAliveDuration int `toml:"iterator-max-alive-duration" json:"iterator-max-alive-duration"`
+
+	// IteratorSlowReadDuration is the iterator slow read threshold.
+	// A reading that exceeds the duration triggers a db compaction.
+	//
+	// The default value is 256, 256ms.
+	IteratorSlowReadDuration int `toml:"iterator-slow-read-duration" json:"iterator-slow-read-duration"`
+
 	// CleanupSpeedLimit limits clean up speed, based on key value entry count.
 	//
 	// The default value is 10000.
diff --git a/pkg/config/server_config.go b/pkg/config/server_config.go
index 4803e8073..dc6890bf2 100644
--- a/pkg/config/server_config.go
+++ b/pkg/config/server_config.go
@@ -120,6 +120,8 @@ var defaultServerConfig = &ServerConfig{
 			WriteL0PauseTrigger:         math.MaxInt32,
 			CompactionL0Trigger:         160,
 			CompactionDeletionThreshold: 160000,
+			IteratorMaxAliveDuration:    10000,
+			IteratorSlowReadDuration:    256,
 			CleanupSpeedLimit:           10000,
 		},
 		Messages: defaultMessageConfig.Clone(),
diff --git a/pkg/db/db.go b/pkg/db/db.go
index c4f54904f..c4f6713cf 100644
--- a/pkg/db/db.go
+++ b/pkg/db/db.go
@@ -15,7 +15,7 @@ package db
 
 // DB is an interface of a leveldb-like database.
 type DB interface {
-	Snapshot() (Snapshot, error)
+	Iterator(lowerBound, upperBound []byte) Iterator
 	Batch(cap int) Batch
 	Compact(start, end []byte) error
 	Close() error
@@ -32,12 +32,6 @@ type Batch interface {
 	Reset()
 }
 
-// Snapshot is an interface of a point-in-time view of the current DB state.
-type Snapshot interface {
-	Iterator(lowerBound, upperBound []byte) Iterator
-	Release() error
-}
-
 // Iterator is an interface of an iterator of a DB.
 type Iterator interface {
 	Valid() bool
diff --git a/pkg/db/leveldb.go b/pkg/db/leveldb.go
index 9618fbfd0..7c53a6fae 100644
--- a/pkg/db/leveldb.go
+++ b/pkg/db/leveldb.go
@@ -86,9 +86,11 @@ type levelDB struct {
 
 var _ DB = (*levelDB)(nil)
 
-func (p *levelDB) Snapshot() (Snapshot, error) {
-	snap, err := p.db.GetSnapshot()
-	return leveldbSnapshot{Snapshot: snap}, err
+func (p *levelDB) Iterator(lowerBound, upperBound []byte) Iterator {
+	return leveldbIter{Iterator: p.db.NewIterator(&util.Range{
+		Start: lowerBound,
+		Limit: upperBound,
+	}, nil)}
 }
 
 func (p *levelDB) Batch(cap int) Batch {
@@ -170,24 +172,6 @@ func (b leveldbBatch) Reset() {
 	b.Batch.Reset()
 }
 
-type leveldbSnapshot struct {
-	*leveldb.Snapshot
-}
-
-var _ Snapshot = (*leveldbSnapshot)(nil)
-
-func (s leveldbSnapshot) Iterator(lowerBound, upperBound []byte) Iterator {
-	return leveldbIter{Iterator: s.NewIterator(&util.Range{
-		Start: lowerBound,
-		Limit: upperBound,
-	}, nil)}
-}
-
-func (s leveldbSnapshot) Release() error {
-	s.Snapshot.Release()
-	return nil
-}
-
 type leveldbIter struct {
 	iterator.Iterator
 }
diff --git a/pkg/db/pebble.go b/pkg/db/pebble.go
index 54ef50ddb..4c458ab33 100644
--- a/pkg/db/pebble.go
+++ b/pkg/db/pebble.go
@@ -138,8 +138,11 @@ type pebbleDB struct {
 
 var _ DB = (*pebbleDB)(nil)
 
-func (p *pebbleDB) Snapshot() (Snapshot, error) {
-	return pebbleSnapshot{Snapshot: p.db.NewSnapshot()}, nil
+func (p *pebbleDB) Iterator(lowerBound, upperBound []byte) Iterator {
+	return pebbleIter{Iterator: p.db.NewIter(&pebble.IterOptions{
+		LowerBound: lowerBound,
+		UpperBound: upperBound,
+	})}
 }
 
 func (p *pebbleDB) Batch(cap int) Batch {
@@ -219,23 +222,6 @@ func (b pebbleBatch) Reset() {
 	b.Batch.Reset()
 }
 
-type pebbleSnapshot struct {
-	*pebble.Snapshot
-}
-
-var _ Snapshot = (*pebbleSnapshot)(nil)
-
-func (s pebbleSnapshot) Iterator(lowerBound, upperBound []byte) Iterator {
-	return pebbleIter{Iterator: s.NewIter(&pebble.IterOptions{
-		LowerBound: lowerBound,
-		UpperBound: upperBound,
-	})}
-}
-
-func (s pebbleSnapshot) Release() error {
-	return s.Close()
-}
-
 type pebbleIter struct {
 	*pebble.Iterator
 }


Test patch:
diff --git a/cdc/sorter/leveldb/cleaner_test.go b/cdc/sorter/leveldb/cleaner_test.go
index 5d638ce34..9ea1915dd 100644
--- a/cdc/sorter/leveldb/cleaner_test.go
+++ b/cdc/sorter/leveldb/cleaner_test.go
@@ -87,66 +87,48 @@ func TestCleanerPoll(t *testing.T) {
 	// Ensure there are some key/values belongs to uid2 table1.
 	start := encoding.EncodeTsKey(2, 1, 0)
 	limit := encoding.EncodeTsKey(2, 2, 0)
-	snap, err := db.Snapshot()
-	require.Nil(t, err)
-	iter := snap.Iterator(start, limit)
+	iter := db.Iterator(start, limit)
 	require.True(t, iter.First())
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Clean up uid2 table1
 	closed := !clean.Poll(ctx, makeCleanTask(2, 1))
 	require.False(t, closed)
 
 	// Ensure no key/values belongs to uid2 table1
-	snap, err = db.Snapshot()
-	require.Nil(t, err)
-	iter = snap.Iterator(start, limit)
+	iter = db.Iterator(start, limit)
 	require.False(t, iter.First())
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Ensure uid1 table1 is untouched.
 	start = encoding.EncodeTsKey(1, 1, 0)
 	limit = encoding.EncodeTsKey(1, 2, 0)
-	snap, err = db.Snapshot()
-	require.Nil(t, err)
-	iter = snap.Iterator(start, limit)
+	iter = db.Iterator(start, limit)
 	require.True(t, iter.First())
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Ensure uid3 table2 is untouched.
 	start = encoding.EncodeTsKey(3, 2, 0)
 	limit = encoding.EncodeTsKey(3, 3, 0)
-	snap, err = db.Snapshot()
-	require.Nil(t, err)
-	iter = snap.Iterator(start, limit)
+	iter = db.Iterator(start, limit)
 	require.True(t, iter.First())
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Clean up uid3 table2
 	closed = !clean.Poll(ctx, makeCleanTask(3, 2))
 	require.False(t, closed)
 
 	// Ensure no key/values belongs to uid3 table2
-	snap, err = db.Snapshot()
-	require.Nil(t, err)
-	iter = snap.Iterator(start, limit)
+	iter = db.Iterator(start, limit)
 	require.False(t, iter.First())
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Ensure uid4 table2 is untouched.
 	start = encoding.EncodeTsKey(4, 2, 0)
 	limit = encoding.EncodeTsKey(4, 3, 0)
-	snap, err = db.Snapshot()
-	require.Nil(t, err)
-	iter = snap.Iterator(start, limit)
+	iter = db.Iterator(start, limit)
 	require.True(t, iter.First())
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Close leveldb.
 	closed = !clean.Poll(ctx, []actormsg.Message{actormsg.StopMessage()})
@@ -207,15 +189,12 @@ func TestCleanerWriteRateLimited(t *testing.T) {
 	keys := [][]byte{}
 	start := encoding.EncodeTsKey(0, 0, 0)
 	limit := encoding.EncodeTsKey(5, 0, 0)
-	snap, err := db.Snapshot()
-	require.Nil(t, err)
-	iter := snap.Iterator(start, limit)
+	iter := db.Iterator(start, limit)
 	for iter.Next() {
 		key := append([]byte{}, iter.Key()...)
 		keys = append(keys, key)
 	}
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 	require.Equal(t, 7, len(keys), "%v", keys)
 
 	// Must speed limited.
@@ -341,12 +320,9 @@ func TestCleanerTaskRescheduled(t *testing.T) {
 	// Ensure all data are deleted.
 	start := encoding.EncodeTsKey(0, 0, 0)
 	limit := encoding.EncodeTsKey(4, 0, 0)
-	snap, err := db.Snapshot()
-	require.Nil(t, err)
-	iter := snap.Iterator(start, limit)
+	iter := db.Iterator(start, limit)
 	require.False(t, iter.First(), fmt.Sprintln(hex.EncodeToString(iter.Key())))
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Close leveldb.
 	closed = !clean.Poll(ctx, []actormsg.Message{actormsg.StopMessage()})
diff --git a/cdc/sorter/leveldb/leveldb_test.go b/cdc/sorter/leveldb/leveldb_test.go
index 53239d94a..b06847e6d 100644
--- a/cdc/sorter/leveldb/leveldb_test.go
+++ b/cdc/sorter/leveldb/leveldb_test.go
@@ -106,28 +106,28 @@ func TestCompact(t *testing.T) {
 	compact.compactThreshold = 2
 
 	// Empty task must not trigger compact.
-	task, snapCh := makeTask(make(map[message.Key][]byte), true)
+	task, iterCh := makeTask(make(map[message.Key][]byte), [][]byte{{0x00}, {0xff}})
 	closed := !ldb.Poll(ctx, task)
 	require.False(t, closed)
-	<-snapCh
+	<-iterCh
 	_, ok := compactMB.Receive()
 	require.False(t, ok)
 
 	// Delete 3 keys must trigger compact.
 	dels := map[message.Key][]byte{"a": {}, "b": {}, "c": {}}
-	task, snapCh = makeTask(dels, true)
+	task, iterCh = makeTask(dels, [][]byte{{0x00}, {0xff}})
 	closed = !ldb.Poll(ctx, task)
 	require.False(t, closed)
-	<-snapCh
+	<-iterCh
 	_, ok = compactMB.Receive()
 	require.True(t, ok)
 
 	// Delete 1 key must not trigger compact.
 	dels = map[message.Key][]byte{"a": {}}
-	task, snapCh = makeTask(dels, true)
+	task, iterCh = makeTask(dels, [][]byte{{0x00}, {0xff}})
 	closed = !ldb.Poll(ctx, task)
 	require.False(t, closed)
-	<-snapCh
+	<-iterCh
 	_, ok = compactMB.Receive()
 	require.False(t, ok)
 
@@ -138,13 +138,20 @@ func TestCompact(t *testing.T) {
 	require.Nil(t, db.Close())
 }
 
-func makeTask(events map[message.Key][]byte, needSnap bool) ([]actormsg.Message, chan message.LimitedSnapshot) {
-	snapCh := make(chan message.LimitedSnapshot, 1)
+func makeTask(events map[message.Key][]byte, rg [][]byte) ([]actormsg.Message, chan *message.LimitedIterator) {
+	var iterReq *message.IterRequest
+	var iterCh chan *message.LimitedIterator
+	if len(rg) != 0 {
+		iterCh = make(chan *message.LimitedIterator, 1)
+		iterReq = &message.IterRequest{
+			Range:  [2][]byte{rg[0], rg[1]},
+			IterCh: iterCh,
+		}
+	}
 	return []actormsg.Message{actormsg.SorterMessage(message.Task{
-		Events:   events,
-		SnapCh:   snapCh,
-		NeedSnap: needSnap,
-	})}, snapCh
+		Events:  events,
+		IterReq: iterReq,
+	})}, iterCh
 }
 
 func TestPutReadDelete(t *testing.T) {
@@ -162,19 +169,18 @@ func TestPutReadDelete(t *testing.T) {
 	require.Nil(t, err)
 
 	// Put only.
-	tasks, snapCh := makeTask(map[message.Key][]byte{"key": {}}, false)
+	tasks, iterCh := makeTask(map[message.Key][]byte{"key": {}}, nil)
+	require.Nil(t, iterCh)
 	closed := !ldb.Poll(ctx, tasks)
 	require.False(t, closed)
-	_, ok := <-snapCh
-	require.False(t, ok)
 
 	// Put and read.
-	tasks, snapCh = makeTask(map[message.Key][]byte{"key": []byte("value")}, true)
+	tasks, iterCh = makeTask(map[message.Key][]byte{"key": []byte("value")},
+		[][]byte{{0x00}, {0xff}})
 	closed = !ldb.Poll(ctx, tasks)
 	require.False(t, closed)
-	snap, ok := <-snapCh
+	iter, ok := <-iterCh
 	require.True(t, ok)
-	iter := snap.Iterator([]byte(""), []byte{0xff})
 	require.NotNil(t, iter)
 	ok = iter.Seek([]byte(""))
 	require.True(t, ok)
@@ -182,15 +188,13 @@ func TestPutReadDelete(t *testing.T) {
 	ok = iter.Next()
 	require.False(t, ok)
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Read only.
-	tasks, snapCh = makeTask(make(map[message.Key][]byte), true)
+	tasks, iterCh = makeTask(make(map[message.Key][]byte), [][]byte{{0x00}, {0xff}})
 	closed = !ldb.Poll(ctx, tasks)
 	require.False(t, closed)
-	snap, ok = <-snapCh
+	iter, ok = <-iterCh
 	require.True(t, ok)
-	iter = snap.Iterator([]byte(""), []byte{0xff})
 	require.NotNil(t, iter)
 	ok = iter.Seek([]byte(""))
 	require.True(t, ok)
@@ -198,20 +202,68 @@ func TestPutReadDelete(t *testing.T) {
 	ok = iter.Next()
 	require.False(t, ok)
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
 
 	// Delete and read.
-	tasks, snapCh = makeTask(map[message.Key][]byte{"key": {}}, true)
+	tasks, iterCh = makeTask(map[message.Key][]byte{"key": {}}, [][]byte{{0x00}, {0xff}})
 	closed = !ldb.Poll(ctx, tasks)
 	require.False(t, closed)
-	snap, ok = <-snapCh
+	iter, ok = <-iterCh
 	require.True(t, ok)
-	iter = snap.Iterator([]byte(""), []byte{0xff})
 	require.NotNil(t, iter)
 	ok = iter.Seek([]byte(""))
 	require.False(t, ok, string(iter.Key()))
 	require.Nil(t, iter.Release())
-	require.Nil(t, snap.Release())
+
+	// Close db.
+	closed = !ldb.Poll(ctx, []actormsg.Message{actormsg.StopMessage()})
+	require.True(t, closed)
+	closedWg.Wait()
+	require.Nil(t, db.Close())
+}
+
+func TestAcquireIterators(t *testing.T) {
+	ctx := context.Background()
+	cfg := config.GetDefaultServerConfig().Clone().Debug.DB
+	cfg.Count = 1
+
+	db, err := db.OpenLevelDB(ctx, 1, t.TempDir(), cfg)
+	require.Nil(t, err)
+	closedWg := new(sync.WaitGroup)
+
+	// Set max iterator count to 1.
+	cfg.Concurrency = 1
+	compact := NewCompactScheduler(actor.NewRouter(t.Name()), cfg)
+	ldb, _, err := NewDBActor(0, db, cfg, compact, closedWg, "")
+	require.Nil(t, err)
+
+	// Poll two tasks.
+	tasks, iterCh1 := makeTask(make(map[message.Key][]byte), [][]byte{{0x00}, {0xff}})
+	tasks[0].SorterTask.TableID = 1
+	tasks2, iterCh2 := makeTask(make(map[message.Key][]byte), [][]byte{{0x00}, {0xff}})
+	tasks2[0].SorterTask.TableID = 2
+	tasks = append(tasks, tasks2...)
+	closed := !ldb.Poll(ctx, tasks)
+	require.False(t, closed)
+	iter, ok := <-iterCh1
+	require.True(t, ok)
+	require.NotNil(t, iter)
+
+	// Require iterator is not allow for now.
+	closed = !ldb.Poll(ctx, []actormsg.Message{actormsg.TickMessage()})
+	require.False(t, closed)
+	select {
+	case <-iterCh2:
+		require.FailNow(t, "should not acquire an iterator")
+	default:
+	}
+
+	// Release iter and iterCh2 should be able to receive an iterator.
+	require.Nil(t, iter.Release())
+	closed = !ldb.Poll(ctx, []actormsg.Message{actormsg.TickMessage()})
+	require.False(t, closed)
+	iter, ok = <-iterCh2
+	require.True(t, ok)
+	require.Nil(t, iter.Release())
 
 	// Close db.
 	closed = !ldb.Poll(ctx, []actormsg.Message{actormsg.StopMessage()})
@@ -288,7 +340,7 @@ func TestModelChecking(t *testing.T) {
 		// Put to model.
 		model.put(message.Key(key), value)
 		// Put to db.
-		tasks, _ := makeTask(map[message.Key][]byte{message.Key(key): value}, false)
+		tasks, _ := makeTask(map[message.Key][]byte{message.Key(key): value}, nil)
 		closed := !ldb.Poll(ctx, tasks)
 		require.False(t, closed)
 	}
@@ -305,7 +357,7 @@ func TestModelChecking(t *testing.T) {
 				value := key
 
 				model.put(message.Key(key), value)
-				tasks, _ := makeTask(map[message.Key][]byte{message.Key(key): value}, false)
+				tasks, _ := makeTask(map[message.Key][]byte{message.Key(key): value}, nil)
 				closed := !ldb.Poll(ctx, tasks)
 				require.False(t, closed)
 
@@ -314,17 +366,17 @@ func TestModelChecking(t *testing.T) {
 				keys := model.iter(minKey, maxKey)
 				delKey := keys[rd.Intn(len(keys))]
 				model.delete(delKey)
-				tasks, _ := makeTask(map[message.Key][]byte{delKey: {}}, false)
+				tasks, _ := makeTask(map[message.Key][]byte{delKey: {}}, nil)
 				closed := !ldb.Poll(ctx, tasks)
 				require.False(t, closed)
 			}
 		}
 
-		tasks, snapCh := makeTask(map[message.Key][]byte{}, true)
+		tasks, iterCh := makeTask(map[message.Key][]byte{},
+			[][]byte{[]byte(minKey), []byte(maxKey)})
 		closed := !ldb.Poll(ctx, tasks)
 		require.False(t, closed)
-		snap := <-snapCh
-		iter := snap.Iterator([]byte(minKey), []byte(maxKey))
+		iter := <-iterCh
 		iter.Seek([]byte(minKey))
 		keys := model.iter(minKey, maxKey)
 		for idx, key := range keys {
@@ -335,7 +387,6 @@ func TestModelChecking(t *testing.T) {
 				"index %d, len(model): %d, seed: %d", idx, len(model.kvs), seed)
 		}
 		require.Nil(t, iter.Release())
-		require.Nil(t, snap.Release())
 	}
 
 	// Close db.
@@ -359,7 +410,7 @@ func TestContextCancel(t *testing.T) {
 	require.Nil(t, err)
 
 	cancel()
-	tasks, _ := makeTask(map[message.Key][]byte{"key": {}}, true)
+	tasks, _ := makeTask(map[message.Key][]byte{"key": {}}, [][]byte{{0x00}, {0xff}})
 	closed := !ldb.Poll(ctx, tasks)
 	require.True(t, closed)
 	closedWg.Wait()
diff --git a/cdc/sorter/leveldb/table_sorter_test.go b/cdc/sorter/leveldb/table_sorter_test.go
index 8d057e85b..75a35afee 100644
--- a/cdc/sorter/leveldb/table_sorter_test.go
+++ b/cdc/sorter/leveldb/table_sorter_test.go
@@ -24,7 +24,6 @@ import (
 	"github.com/pingcap/tiflow/cdc/sorter/encoding"
 	"github.com/pingcap/tiflow/cdc/sorter/leveldb/message"
 	"github.com/pingcap/tiflow/pkg/actor"
-	actormsg "github.com/pingcap/tiflow/pkg/actor/message"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/db"
 	"github.com/stretchr/testify/require"
@@ -209,7 +208,7 @@ func TestWaitOutput(t *testing.T) {
 	require.EqualValues(t, 0, rts)
 }
 
-func TestAsyncWrite(t *testing.T) {
+func TestBuildTask(t *testing.T) {
 	t.Parallel()
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -217,18 +216,16 @@ func TestAsyncWrite(t *testing.T) {
 
 	capacity := 4
 	require.Greater(t, batchReceiveEventSize, capacity)
-	ls, mb := newTestSorter(ctx, capacity)
+	ls, _ := newTestSorter(ctx, capacity)
 
 	cases := []struct {
 		events     []*model.PolymorphicEvent
 		deleteKeys []message.Key
-		needSnap   bool
 	}{
 		// Empty write and delete.
 		{
 			events:     []*model.PolymorphicEvent{},
 			deleteKeys: []message.Key{},
-			needSnap:   false,
 		},
 		// Write one event
 		{
@@ -236,7 +233,6 @@ func TestAsyncWrite(t *testing.T) {
 				model.NewPolymorphicEvent(&model.RawKVEntry{CRTs: 1}),
 			},
 			deleteKeys: []message.Key{},
-			needSnap:   false,
 		},
 		// Write one event and delete one key.
 		{
@@ -247,18 +243,6 @@ func TestAsyncWrite(t *testing.T) {
 				message.Key(encoding.EncodeKey(ls.uid, ls.tableID,
 					model.NewPolymorphicEvent(&model.RawKVEntry{CRTs: 2}))),
 			},
-			needSnap: false,
-		},
-		// Write one event and delete one key and need iter.
-		{
-			events: []*model.PolymorphicEvent{
-				model.NewPolymorphicEvent(&model.RawKVEntry{CRTs: 1}),
-			},
-			deleteKeys: []message.Key{
-				message.Key(encoding.EncodeKey(ls.uid, ls.tableID,
-					model.NewPolymorphicEvent(&model.RawKVEntry{CRTs: 2}))),
-			},
-			needSnap: true,
 		},
 		// Write one event and a resolved ts event.
 		{
@@ -267,7 +251,6 @@ func TestAsyncWrite(t *testing.T) {
 				model.NewResolvedPolymorphicEvent(0, 3),
 			},
 			deleteKeys: []message.Key{},
-			needSnap:   false,
 		},
 		// Write two events and a resolved ts event, delete one key.
 		{
@@ -280,17 +263,12 @@ func TestAsyncWrite(t *testing.T) {
 				message.Key(encoding.EncodeKey(ls.uid, ls.tableID,
 					model.NewPolymorphicEvent(&model.RawKVEntry{CRTs: 1}))),
 			},
-			needSnap: false,
 		},
 	}
 	for i, cs := range cases {
-		events, deleteKeys, needSnap := cs.events, cs.deleteKeys, cs.needSnap
-		ch, err := ls.asyncWrite(ctx, events, deleteKeys, needSnap)
+		events, deleteKeys := cs.events, cs.deleteKeys
+		task, err := ls.buildTask(events, deleteKeys)
 		require.Nil(t, err, "case #%d, %v", i, cs)
-		require.NotNil(t, ch, "case #%d, %v", i, cs)
-		task, ok := mb.Receive()
-		require.True(t, ok, "case #%d, %v", i, cs)
-		require.EqualValues(t, actormsg.TypeSorterTask, task.Tp)
 
 		expectedEvents := make(map[message.Key][]uint8)
 		for _, ev := range events {
@@ -310,11 +288,9 @@ func TestAsyncWrite(t *testing.T) {
 			UID:                ls.uid,
 			TableID:            ls.tableID,
 			Events:             expectedEvents,
-			SnapCh:             ch,
-			NeedSnap:           needSnap,
 			Cleanup:            false,
 			CleanupRatelimited: false,
-		}, task.SorterTask, "case #%d, %v", i, cs)
+		}, task, "case #%d, %v", i, cs)
 	}
 }
 
@@ -698,11 +674,10 @@ func TestOutputIterEvents(t *testing.T) {
 		ls.outputCh = make(chan *model.PolymorphicEvent, cs.outputChCap)
 		buf := newOutputBuffer(capacity)
 
-		snap, err := db.Snapshot()
-		require.Nil(t, err, "case #%d, %v", i, cs)
-		iter := snap.Iterator(
+		iter := db.Iterator(
 			encoding.EncodeTsKey(ls.uid, ls.tableID, 0),
 			encoding.EncodeTsKey(ls.uid, ls.tableID, cs.maxResolvedTs+1))
+		require.Nil(t, iter.Error(), "case #%d, %v", i, cs)
 		exhaustedRTs, err := ls.outputIterEvents(iter, buf, cs.maxResolvedTs)
 		require.Nil(t, err, "case #%d, %v", i, cs)
 		require.EqualValues(t, cs.expectExhaustedRTs, exhaustedRTs, "case #%d, %v", i, cs)
@@ -741,7 +716,7 @@ func TestPoll(t *testing.T) {
 
 	cases := []struct {
 		inputEvents []*model.PolymorphicEvent
-		inputSnap   func() *message.LimitedSnapshot
+		inputIter   func([2][]byte) *message.LimitedIterator
 		state       *pollState
 
 		expectEvents        []*model.PolymorphicEvent
@@ -759,7 +734,7 @@ func TestPoll(t *testing.T) {
 				eventsBuf: make([]*model.PolymorphicEvent, 1),
 				outputBuf: newOutputBuffer(1),
 			},
-			inputSnap: func() *message.LimitedSnapshot { return nil },
+			inputIter: func([2][]byte) *message.LimitedIterator { return nil },
 
 			expectEvents:     []*model.PolymorphicEvent{},
 			expectDeleteKeys: []message.Key{},
@@ -780,8 +755,8 @@ func TestPoll(t *testing.T) {
 				eventsBuf: make([]*model.PolymorphicEvent, 2),
 				outputBuf: newOutputBuffer(1),
 			},
-			// An empty snapshot.
-			inputSnap: newEmptySnapshot(ctx, t, sema),
+			// An empty iterator.
+			inputIter: newEmptyIterator(ctx, t, db, sema),
 
 			expectEvents:     []*model.PolymorphicEvent{},
 			expectDeleteKeys: []message.Key{},
@@ -803,7 +778,7 @@ func TestPoll(t *testing.T) {
 				eventsBuf: make([]*model.PolymorphicEvent, 2),
 				outputBuf: newOutputBuffer(1),
 			},
-			inputSnap: newSnapshot(ctx, t, db, sema),
+			inputIter: newSnapshot(ctx, t, db, sema),
 
 			expectEvents: []*model.PolymorphicEvent{},
 			expectDeleteKeys: []message.Key{
@@ -832,7 +807,7 @@ func TestPoll(t *testing.T) {
 				maxCommitTs:         2,
 				exhaustedResolvedTs: 2,
 			},
-			inputSnap: func() *message.LimitedSnapshot { return nil },
+			inputIter: func([2][]byte) *message.LimitedIterator { return nil },
 
 			expectEvents:     []*model.PolymorphicEvent{},
 			expectDeleteKeys: []message.Key{},
@@ -858,7 +833,7 @@ func TestPoll(t *testing.T) {
 					advisedCapacity: 2,
 				},
 			},
-			inputSnap: func() *message.LimitedSnapshot { return nil },
+			inputIter: func([2][]byte) *message.LimitedIterator { return nil },
 
 			expectEvents:     []*model.PolymorphicEvent{},
 			expectDeleteKeys: []message.Key{},
@@ -879,7 +854,7 @@ func TestPoll(t *testing.T) {
 		_, state := i, cs.state
 		var wg sync.WaitGroup
 		wg.Add(1)
-		go func() { handleTask(mb, cs.inputSnap(), &wg) }()
+		go func() { handleTask(mb, &wg, cs.inputIter) }()
 		for i := range cs.inputEvents {
 			ls.AddEntry(ctx, cs.inputEvents[i])
 		}
@@ -899,7 +874,10 @@ func TestPoll(t *testing.T) {
 	require.Nil(t, db.Close())
 }
 
-func handleTask(mb actor.Mailbox, iter *message.LimitedSnapshot, wg *sync.WaitGroup) {
+func handleTask(
+	mb actor.Mailbox, wg *sync.WaitGroup,
+	iterFn func(rg [2][]byte) *message.LimitedIterator,
+) {
 	defer wg.Done()
 	for {
 		time.Sleep(20 * time.Millisecond)
@@ -907,56 +885,37 @@ func handleTask(mb actor.Mailbox, iter *message.LimitedSnapshot, wg *sync.WaitGr
 		if !ok {
 			continue
 		}
+		if task.SorterTask.IterReq == nil {
+			return
+		}
+		iter := iterFn(task.SorterTask.IterReq.Range)
 		if iter != nil {
-			task.SorterTask.SnapCh <- *iter
+			task.SorterTask.IterReq.IterCh <- iter
 		}
-		close(task.SorterTask.SnapCh)
+		close(task.SorterTask.IterReq.IterCh)
 		return
 	}
 }
 
 func newSnapshot(
-	ctx context.Context, t *testing.T, db db.DB,
-	sema *semaphore.Weighted,
-) func() *message.LimitedSnapshot {
-	return func() *message.LimitedSnapshot {
+	ctx context.Context, t *testing.T, db db.DB, sema *semaphore.Weighted,
+) func(rg [2][]byte) *message.LimitedIterator {
+	return func(rg [2][]byte) *message.LimitedIterator {
 		require.Nil(t, sema.Acquire(ctx, 1))
-		snap, err := db.Snapshot()
-		require.Nil(t, err)
-		return &message.LimitedSnapshot{
-			Snapshot: snap,
+		return &message.LimitedIterator{
+			Iterator: db.Iterator(rg[0], rg[1]),
 			Sema:     sema,
 		}
 	}
 }
 
-type emptySnapshot struct {
-	db.Snapshot
-}
-
-func (e emptySnapshot) Iterator(_, _ []byte) db.Iterator { return emptyIterator{} }
-func (e emptySnapshot) Release() error                   { return nil }
-
-type emptyIterator struct {
-	db.Iterator
-}
-
-func (emptyIterator) Valid() bool      { return false }
-func (emptyIterator) First() bool      { return false }
-func (emptyIterator) Seek([]byte) bool { return false }
-func (emptyIterator) Next() bool       { return false }
-func (emptyIterator) Key() []byte      { return nil }
-func (emptyIterator) Value() []byte    { return nil }
-func (emptyIterator) Error() error     { return nil }
-func (emptyIterator) Release() error   { return nil }
-
-func newEmptySnapshot(
-	ctx context.Context, t *testing.T, sema *semaphore.Weighted,
-) func() *message.LimitedSnapshot {
-	return func() *message.LimitedSnapshot {
+func newEmptyIterator(
+	ctx context.Context, t *testing.T, db db.DB, sema *semaphore.Weighted,
+) func(rg [2][]byte) *message.LimitedIterator {
+	return func(rg [2][]byte) *message.LimitedIterator {
 		require.Nil(t, sema.Acquire(ctx, 1))
-		return &message.LimitedSnapshot{
-			Snapshot: emptySnapshot{},
+		return &message.LimitedIterator{
+			Iterator: db.Iterator([]byte{}, []byte{}),
 			Sema:     sema,
 		}
 	}
diff --git a/pkg/cmd/server/server_test.go b/pkg/cmd/server/server_test.go
index b77d4a59a..1a36d1cc0 100644
--- a/pkg/cmd/server/server_test.go
+++ b/pkg/cmd/server/server_test.go
@@ -192,6 +192,8 @@ func TestParseCfg(t *testing.T) {
 				WriteL0PauseTrigger:         math.MaxInt32,
 				CompactionL0Trigger:         160,
 				CompactionDeletionThreshold: 160000,
+				IteratorMaxAliveDuration:    10000,
+				IteratorSlowReadDuration:    256,
 				CleanupSpeedLimit:           10000,
 			},
 			// We expect the default configuration here.
@@ -329,6 +331,8 @@ server-worker-pool-size = 16
 				WriteL0SlowdownTrigger:      12,
 				WriteL0PauseTrigger:         13,
 				CleanupSpeedLimit:           14,
+				IteratorMaxAliveDuration:    10000,
+				IteratorSlowReadDuration:    256,
 				CompactionDeletionThreshold: 15,
 			},
 			Messages: &config.MessagesConfig{
@@ -464,6 +468,8 @@ cert-allowed-cn = ["dd","ee"]
 				WriteL0PauseTrigger:         math.MaxInt32,
 				CompactionL0Trigger:         160,
 				CompactionDeletionThreshold: 160000,
+				IteratorMaxAliveDuration:    10000,
+				IteratorSlowReadDuration:    256,
 				CleanupSpeedLimit:           10000,
 			},
 			// We expect the default configuration here.
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index dbffe4ec5..68a2d0459 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -123,6 +123,8 @@ const (
       "write-l0-pause-trigger": 2147483647,
       "compaction-l0-trigger": 160,
       "compaction-deletion-threshold": 160000,
+      "iterator-max-alive-duration": 10000,
+      "iterator-slow-read-duration": 256,
       "cleanup-speed-limit": 10000
     },
     "messages": {
diff --git a/pkg/db/db_test.go b/pkg/db/db_test.go
index c6a84de83..d6edcd04c 100644
--- a/pkg/db/db_test.go
+++ b/pkg/db/db_test.go
@@ -75,18 +75,12 @@ func testDB(t *testing.T, db DB) {
 	batch.Reset()
 	require.EqualValues(t, lbatch.Len(), batch.Count())
 
-	// Snapshot
-	lsnap, err := ldb.GetSnapshot()
-	require.Nil(t, err)
-	snap, err := db.Snapshot()
-	require.Nil(t, err)
-
 	// Iterator
-	liter := lsnap.NewIterator(&util.Range{
+	liter := ldb.NewIterator(&util.Range{
 		Start: []byte(""),
 		Limit: []byte("k4"),
 	}, nil)
-	iter := snap.Iterator([]byte(""), []byte("k4"))
+	iter := db.Iterator([]byte(""), []byte("k4"))
 	// First
 	require.True(t, liter.First())
 	require.True(t, iter.First())
@@ -115,8 +109,6 @@ func testDB(t *testing.T, db DB) {
 	// Release
 	liter.Release()
 	require.Nil(t, iter.Release())
-	lsnap.Release()
-	require.Nil(t, snap.Release())
 
 	// Compact
 	require.Nil(t, db.Compact([]byte{0x00}, []byte{0xff}))
@@ -248,9 +240,8 @@ func BenchmarkNext(b *testing.B) {
 				b.ResetTimer()
 
 				b.Run(fmt.Sprintf("next %d event(s)", count), func(b *testing.B) {
-					snap, err := db.Snapshot()
-					require.Nil(b, err)
-					iter := snap.Iterator([]byte{}, bytes.Repeat([]byte{0xff}, len(key)))
+					iter := db.Iterator([]byte{}, bytes.Repeat([]byte{0xff}, len(key)))
+					require.Nil(b, iter.Error())
 					for i := 0; i < b.N; i++ {
 						for ok := iter.First(); ok; ok = iter.Next() {
 						}