Repository: matrixorigin/matrixone
Pull request #13892 (2023-12-28 07:13:15)

CI: change runner for race bvt
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:


Patch:
diff --git a/.github/workflows/merge-trigger-standalone.yaml b/.github/workflows/merge-trigger-standalone.yaml
index d475734c2..069677a67 100644
--- a/.github/workflows/merge-trigger-standalone.yaml
+++ b/.github/workflows/merge-trigger-standalone.yaml
@@ -110,7 +110,7 @@ jobs:
 
   multi-cn-bvt-race-linux-x86:
     if: ${{ github.event.pull_request.merged == true && always() }}
-    runs-on: amd64-shanghai-4xlarge32
+    runs-on: amd64-mo-guangzhou-2xlarge32
     name: Multi-CN e2e BVT(Race) Test on Linux/x64
     needs: [pessimistic-bvt-darwin-arm64]
     steps:
@@ -209,7 +209,3 @@ jobs:
         with:
           msgtype: markdown
           content: "MO Checkin Regression(Standalone BVT) failed: [${{ github.event.pull_request.title }}](${{ github.event.pull_request.html_url }})\nCreator:<font color=\"warning\">${{ github.event.pull_request.user.login }}</font>\nAction Link: [Link](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})<@shuqi.qin><@JiejieJia><@SuDong>"
-
-      - name: SlEEP FOR CHECK
-        if: ${{ always() && steps.check_mo_status.conclusion != 'success' }}
-        run: sleep 21600
diff --git a/pkg/common/morpc/future.go b/pkg/common/morpc/future.go
index 17e02d42a..1031479d2 100644
--- a/pkg/common/morpc/future.go
+++ b/pkg/common/morpc/future.go
@@ -42,6 +42,7 @@ type Future struct {
 	writtenC    chan error
 	waiting     atomic.Bool
 	releaseFunc func(*Future)
+	oneWay      bool
 	mu          struct {
 		sync.Mutex
 		notified bool
@@ -52,13 +53,14 @@ type Future struct {
 }
 
 func (f *Future) init(send RPCMessage) {
-	if _, ok := send.Ctx.Deadline(); !ok && !send.internal {
+	if _, ok := send.Ctx.Deadline(); !ok && !send.oneWay && !send.internal {
 		panic("context deadline not set")
 	}
 	f.waiting.Store(false)
 	f.send = send
 	f.send.createAt = time.Now()
 	f.id = send.Message.GetID()
+	f.oneWay = send.oneWay
 	f.mu.Lock()
 	f.mu.closed = false
 	f.mu.notified = false
@@ -98,11 +100,14 @@ func (f *Future) Close() {
 }
 
 func (f *Future) waitSendCompleted() error {
+	if f.oneWay {
+		panic("one way cannot call waitSendCompleted")
+	}
 	return <-f.writtenC
 }
 
 func (f *Future) messageSent(err error) {
-	if f.waiting.CompareAndSwap(false, true) {
+	if !f.oneWay && f.waiting.CompareAndSwap(false, true) {
 		f.writtenC <- err
 		f.unRef()
 	}
diff --git a/pkg/common/morpc/message.go b/pkg/common/morpc/message.go
index 4a7be59c7..7a2cf743e 100644
--- a/pkg/common/morpc/message.go
+++ b/pkg/common/morpc/message.go
@@ -22,6 +22,7 @@ import (
 
 const (
 	internalTimeout = time.Second * 10
+	oneWayTimeout   = time.Second * 600
 )
 
 // Timeout return true if the message is timeout
@@ -39,6 +40,9 @@ func (m RPCMessage) GetTimeoutFromContext() (time.Duration, error) {
 	if m.internal {
 		return internalTimeout, nil
 	}
+	if m.oneWay {
+		return oneWayTimeout, nil
+	}
 
 	d, ok := m.Ctx.Deadline()
 	if !ok {
diff --git a/pkg/common/morpc/mock_morpc/types_mock.go b/pkg/common/morpc/mock_morpc/types_mock.go
index aa26719af..aef0ea272 100644
--- a/pkg/common/morpc/mock_morpc/types_mock.go
+++ b/pkg/common/morpc/mock_morpc/types_mock.go
@@ -127,6 +127,11 @@ func (m *MockClientSession) Write(arg0 context.Context, arg1 morpc.Message) erro
 	return ret0
 }
 
+// AsyncWrite mocks base method.
+func (m *MockClientSession) AsyncWrite(arg1 morpc.Message) error {
+	panic("not supported")
+}
+
 // Write indicates an expected call of Write.
 func (mr *MockClientSessionMockRecorder) Write(arg0, arg1 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
diff --git a/pkg/common/morpc/server.go b/pkg/common/morpc/server.go
index 9f8d219b6..ce4d547d1 100644
--- a/pkg/common/morpc/server.go
+++ b/pkg/common/morpc/server.go
@@ -281,6 +281,7 @@ func (s *server) startWriteLoop(cs *clientSession) error {
 		defer s.closeClientSession(cs)
 
 		responses := make([]*Future, 0, s.options.batchSendSize)
+		needClose := make([]*Future, 0, s.options.batchSendSize)
 		fetch := func() {
 			defer func() {
 				cs.metrics.sendingQueueSizeGauge.Set(float64(len(cs.c)))
@@ -289,7 +290,11 @@ func (s *server) startWriteLoop(cs *clientSession) error {
 			for i := 0; i < len(responses); i++ {
 				responses[i] = nil
 			}
+			for i := 0; i < len(needClose); i++ {
+				needClose[i] = nil
+			}
 			responses = responses[:0]
+			needClose = needClose[:0]
 
 			for i := 0; i < s.options.batchSendSize; i++ {
 				if len(responses) == 0 {
@@ -346,6 +351,9 @@ func (s *server) startWriteLoop(cs *clientSession) error {
 					timeout := time.Duration(0)
 					for _, f := range responses {
 						s.metrics.writeLatencyDurationHistogram.Observe(start.Sub(f.send.createAt).Seconds())
+						if f.oneWay {
+							needClose = append(needClose, f)
+						}
 
 						if !s.options.filter(f.send.Message) {
 							f.messageSent(messageSkipped)
@@ -410,6 +418,9 @@ func (s *server) startWriteLoop(cs *clientSession) error {
 					for _, f := range responses {
 						f.messageSent(nil)
 					}
+					for _, f := range needClose {
+						f.Close()
+					}
 
 					s.metrics.writeDurationHistogram.Observe(time.Since(start).Seconds())
 				}
@@ -519,7 +530,7 @@ func newClientSession(
 		metrics:                 metrics,
 		closedC:                 make(chan struct{}),
 		codec:                   codec,
-		c:                       make(chan *Future, 32),
+		c:                       make(chan *Future, 1024),
 		receivedStreamSequences: make(map[uint64]uint32),
 		conn:                    conn,
 		ctx:                     ctx,
@@ -589,6 +600,15 @@ func (cs *clientSession) Write(
 	})
 }
 
+func (cs *clientSession) AsyncWrite(response Message) error {
+	_, err := cs.send(RPCMessage{
+		Ctx:     context.Background(),
+		Message: response,
+		oneWay:  true,
+	})
+	return err
+}
+
 func (cs *clientSession) send(msg RPCMessage) (*Future, error) {
 	cs.metrics.sendCounter.Inc()
 
@@ -613,8 +633,10 @@ func (cs *clientSession) send(msg RPCMessage) (*Future, error) {
 	}
 
 	f := cs.newFutureFunc()
-	f.ref()
 	f.init(msg)
+	if !f.oneWay {
+		f.ref()
+	}
 	cs.c <- f
 	cs.metrics.sendingQueueSizeGauge.Set(float64(len(cs.c)))
 	return f, nil
diff --git a/pkg/common/morpc/types.go b/pkg/common/morpc/types.go
index 6649768eb..ed357750d 100644
--- a/pkg/common/morpc/types.go
+++ b/pkg/common/morpc/types.go
@@ -68,6 +68,7 @@ type RPCMessage struct {
 	// Message raw rpc message
 	Message Message
 
+	oneWay         bool
 	internal       bool
 	stream         bool
 	streamSequence uint32
@@ -104,6 +105,8 @@ type ClientSession interface {
 	Close() error
 	// Write writing the response message to the client.
 	Write(ctx context.Context, response Message) error
+	// AsyncWrite only put message into write queue, and return immediately.
+	AsyncWrite(response Message) error
 	// CreateCache create a message cache using cache ID. Cache will removed if
 	// context is done.
 	CreateCache(ctx context.Context, cacheID uint64) (MessageCache, error)
diff --git a/pkg/lockservice/cfg.go b/pkg/lockservice/cfg.go
index 21c67f55a..d57fdf356 100644
--- a/pkg/lockservice/cfg.go
+++ b/pkg/lockservice/cfg.go
@@ -62,6 +62,9 @@ type Config struct {
 	// continuously hold the bind, and if no hold request is received after the configured time,
 	// then all bindings for the service will fail.
 	KeepBindTimeout toml.Duration `toml:"keep-bind-timeout"`
+	// EnableRemoteLocalProxy enable remote local proxy. The proxy used to reduce remote shared
+	// lock and unlock request.
+	EnableRemoteLocalProxy bool `toml:"enable-remote-local-proxy"`
 }
 
 // Validate validate
diff --git a/pkg/lockservice/lock.go b/pkg/lockservice/lock.go
index 762b69340..1521bfc7f 100644
--- a/pkg/lockservice/lock.go
+++ b/pkg/lockservice/lock.go
@@ -283,6 +283,22 @@ func (h *holders) remove(txnID []byte) {
 	h.txns = newTxns
 }
 
+func (h *holders) replace(
+	from []byte,
+	to pb.WaitTxn) {
+	find := false
+	for i := range h.txns {
+		if bytes.Equal(h.txns[i].TxnID, from) {
+			h.txns[i] = to
+			find = true
+			continue
+		}
+	}
+	if !find {
+		panic("BUG: missing holder")
+	}
+}
+
 func (h *holders) clear() {
 	h.txns = h.txns[:0]
 }
diff --git a/pkg/lockservice/lock_table_allocator.go b/pkg/lockservice/lock_table_allocator.go
index 81caeed58..17e644db0 100644
--- a/pkg/lockservice/lock_table_allocator.go
+++ b/pkg/lockservice/lock_table_allocator.go
@@ -37,7 +37,7 @@ type lockTableAllocator struct {
 	mu struct {
 		sync.RWMutex
 		services   map[string]*serviceBinds
-		lockTables map[uint64]pb.LockTable
+		lockTables map[uint32]map[uint64]pb.LockTable
 	}
 }
 
@@ -59,7 +59,7 @@ func NewLockTableAllocator(
 			stopper.WithLogger(logger.RawLogger().Named(tag))),
 		keepBindTimeout: keepBindTimeout,
 	}
-	la.mu.lockTables = make(map[uint64]pb.LockTable, 10240)
+	la.mu.lockTables = make(map[uint32]map[uint64]pb.LockTable)
 	la.mu.services = make(map[string]*serviceBinds)
 	if err := la.stopper.RunTask(la.checkInvalidBinds); err != nil {
 		panic(err)
@@ -71,13 +71,16 @@ func NewLockTableAllocator(
 
 func (l *lockTableAllocator) Get(
 	serviceID string,
-	tableID uint64) pb.LockTable {
+	group uint32,
+	tableID uint64,
+	originTableID uint64,
+	sharding pb.Sharding) pb.LockTable {
 	binds := l.getServiceBinds(serviceID)
 	if binds == nil {
 		binds = l.registerService(serviceID, tableID)
 	}
 	binds.active()
-	return l.registerBind(binds, tableID)
+	return l.registerBind(binds, group, tableID, originTableID, sharding)
 }
 
 func (l *lockTableAllocator) KeepLockTableBind(serviceID string) bool {
@@ -97,21 +100,7 @@ func (l *lockTableAllocator) Valid(binds []pb.LockTable) []uint64 {
 			panic("BUG")
 		}
 
-		// For upgrade, we must abort all old cn version's transactions.
-		// Because the previous version does not support shared Table
-		// segregation by tenant, then during the upgrade process there will
-		// be 2 versions of CN at the same time, resulting in WW conflicts
-		// that may be missed to be detected.
-		// FIXME(fagongzi): remove this logic in next version.
-		if isSharedTable(b.Table) {
-			_, _, ok := decodeSharedTableID(b.Table)
-			if !ok {
-				invalid = append(invalid, b.Table)
-				continue
-			}
-		}
-
-		current, ok := l.mu.lockTables[b.Table]
+		current, ok := l.getLockTablesLocked(b.Group)[b.Table]
 		if !ok ||
 			current.Changed(b) {
 			if l.logger.Enabled(zap.DebugLevel) {
@@ -138,13 +127,16 @@ func (l *lockTableAllocator) disableTableBinds(b *serviceBinds) {
 	defer l.mu.Unlock()
 	// we can't just delete the LockTable's effectiveness binding directly, we
 	// need to keep the binding version.
-	for table := range b.tables {
-		if old, ok := l.mu.lockTables[table]; ok &&
-			old.ServiceID == b.serviceID {
-			old.Valid = false
-			l.mu.lockTables[table] = old
+	for g, tables := range b.groupTables {
+		for table := range tables {
+			if old, ok := l.getLockTablesLocked(g)[table]; ok &&
+				old.ServiceID == b.serviceID {
+				old.Valid = false
+				l.getLockTablesLocked(g)[table] = old
+			}
 		}
 	}
+
 	// service need deleted, because this service may never restart
 	delete(l.mu.services, b.serviceID)
 	l.logger.Debug("lock service disabled",
@@ -192,18 +184,22 @@ func (l *lockTableAllocator) registerService(
 
 func (l *lockTableAllocator) registerBind(
 	binds *serviceBinds,
-	tableID uint64) pb.LockTable {
+	group uint32,
+	tableID uint64,
+	originTableID uint64,
+	sharding pb.Sharding) pb.LockTable {
 	l.mu.Lock()
 	defer l.mu.Unlock()
 
-	if old, ok := l.mu.lockTables[tableID]; ok {
-		return l.tryRebindLocked(binds, old, tableID)
+	if old, ok := l.getLockTablesLocked(group)[tableID]; ok {
+		return l.tryRebindLocked(binds, group, old, tableID)
 	}
-	return l.createBindLocked(binds, tableID)
+	return l.createBindLocked(binds, group, tableID, originTableID, sharding)
 }
 
 func (l *lockTableAllocator) tryRebindLocked(
 	binds *serviceBinds,
+	group uint32,
 	old pb.LockTable,
 	tableID uint64) pb.LockTable {
 	// find a valid table and service bind
@@ -214,7 +210,7 @@ func (l *lockTableAllocator) tryRebindLocked(
 	// been invalidated, and the current service has also been invalidated, so
 	// there is no need for any re-bind operation here, and the invalid bind
 	// information is directly returned to the service.
-	if !binds.bind(tableID) {
+	if !binds.bind(group, tableID) {
 		return old
 	}
 
@@ -222,26 +218,39 @@ func (l *lockTableAllocator) tryRebindLocked(
 	old.ServiceID = binds.serviceID
 	old.Version++
 	old.Valid = true
-	l.mu.lockTables[tableID] = old
+	l.getLockTablesLocked(group)[tableID] = old
 	return old
 }
 
 func (l *lockTableAllocator) createBindLocked(
 	binds *serviceBinds,
-	tableID uint64) pb.LockTable {
+	group uint32,
+	tableID uint64,
+	originTableID uint64,
+	sharding pb.Sharding) pb.LockTable {
 	// current service is invalid
-	if !binds.bind(tableID) {
+	if !binds.bind(group, tableID) {
 		return pb.LockTable{}
 	}
 
+	if originTableID == 0 {
+		if sharding == pb.Sharding_ByRow {
+			panic("invalid sharding origin table id")
+		}
+		originTableID = tableID
+	}
+
 	// create new table and service bind
 	b := pb.LockTable{
-		Table:     tableID,
-		ServiceID: binds.serviceID,
-		Version:   1,
-		Valid:     true,
+		Table:       tableID,
+		OriginTable: originTableID,
+		ServiceID:   binds.serviceID,
+		Version:     1,
+		Valid:       true,
+		Sharding:    sharding,
+		Group:       group,
 	}
-	l.mu.lockTables[tableID] = b
+	l.getLockTablesLocked(group)[tableID] = b
 	return b
 }
 
@@ -272,7 +281,7 @@ type serviceBinds struct {
 	sync.RWMutex
 	logger            *log.MOLogger
 	serviceID         string
-	tables            map[uint64]struct{}
+	groupTables       map[uint32]map[uint64]struct{}
 	lastKeepaliveTime time.Time
 	disabled          bool
 }
@@ -283,7 +292,7 @@ func newServiceBinds(
 	return &serviceBinds{
 		serviceID:         serviceID,
 		logger:            logger,
-		tables:            make(map[uint64]struct{}, 1024),
+		groupTables:       make(map[uint32]map[uint64]struct{}),
 		lastKeepaliveTime: time.Now(),
 	}
 }
@@ -295,18 +304,20 @@ func (b *serviceBinds) active() bool {
 		return false
 	}
 	b.lastKeepaliveTime = time.Now()
-	b.logger.Debug("lock service binds actived")
+	b.logger.Debug("lock service binds active")
 	return true
 }
 
-func (b *serviceBinds) bind(tableID uint64) bool {
+func (b *serviceBinds) bind(
+	group uint32,
+	tableID uint64) bool {
 	b.Lock()
 	defer b.Unlock()
 	if b.disabled {
 		return false
 	}
-	b.tables[tableID] = struct{}{}
-	b.logger.Debug("table binded",
+	b.getTablesLocked(group)[tableID] = struct{}{}
+	b.logger.Debug("table bind added",
 		zap.Uint64("table", tableID))
 	return true
 }
@@ -327,6 +338,16 @@ func (b *serviceBinds) disable() {
 	b.logger.Debug("lock service binds disabled")
 }
 
+func (b *serviceBinds) getTablesLocked(group uint32) map[uint64]struct{} {
+	m, ok := b.groupTables[group]
+	if ok {
+		return m
+	}
+	m = make(map[uint64]struct{}, 1024)
+	b.groupTables[group] = m
+	return m
+}
+
 func (l *lockTableAllocator) initServer(cfg morpc.Config) {
 	s, err := NewServer(l.address, cfg)
 	if err != nil {
@@ -357,8 +378,12 @@ func (l *lockTableAllocator) handleGetBind(
 	req *pb.Request,
 	resp *pb.Response,
 	cs morpc.ClientSession) {
-	resp.GetBind.LockTable = l.Get(req.GetBind.ServiceID,
-		req.GetBind.Table)
+	resp.GetBind.LockTable = l.Get(
+		req.GetBind.ServiceID,
+		req.GetBind.Group,
+		req.GetBind.Table,
+		req.GetBind.OriginTable,
+		req.GetBind.Sharding)
 	writeResponse(ctx, cancel, resp, nil, cs)
 }
 
@@ -371,3 +396,13 @@ func (l *lockTableAllocator) handleKeepLockTableBind(
 	resp.KeepLockTableBind.OK = l.KeepLockTableBind(req.KeepLockTableBind.ServiceID)
 	writeResponse(ctx, cancel, resp, nil, cs)
 }
+
+func (l *lockTableAllocator) getLockTablesLocked(group uint32) map[uint64]pb.LockTable {
+	m, ok := l.mu.lockTables[group]
+	if ok {
+		return m
+	}
+	m = make(map[uint64]pb.LockTable, 10240)
+	l.mu.lockTables[group] = m
+	return m
+}
diff --git a/pkg/lockservice/lock_table_keeper.go b/pkg/lockservice/lock_table_keeper.go
index f4b65f449..1e3704527 100644
--- a/pkg/lockservice/lock_table_keeper.go
+++ b/pkg/lockservice/lock_table_keeper.go
@@ -30,7 +30,7 @@ type lockTableKeeper struct {
 	stopper                   *stopper.Stopper
 	keepLockTableBindInterval time.Duration
 	keepRemoteLockInterval    time.Duration
-	tables                    *sync.Map
+	groupTables               *sync.Map // group -> sync.map -> table id -> locktable
 	canDoKeep                 bool
 }
 
@@ -42,11 +42,11 @@ func NewLockTableKeeper(
 	client Client,
 	keepLockTableBindInterval time.Duration,
 	keepRemoteLockInterval time.Duration,
-	tables *sync.Map) LockTableKeeper {
+	groupTables *sync.Map) LockTableKeeper {
 	s := &lockTableKeeper{
 		serviceID:                 serviceID,
 		client:                    client,
-		tables:                    tables,
+		groupTables:               groupTables,
 		keepLockTableBindInterval: keepLockTableBindInterval,
 		keepRemoteLockInterval:    keepRemoteLockInterval,
 		stopper: stopper.NewStopper("lock-table-keeper",
@@ -118,12 +118,16 @@ func (k *lockTableKeeper) doKeepRemoteLock(
 	binds = binds[:0]
 	futures = futures[:0]
 
-	k.tables.Range(func(key, value any) bool {
-		lb := value.(lockTable)
-		bind := lb.getBind()
-		if bind.ServiceID != k.serviceID {
-			services[bind.ServiceID] = bind
-		}
+	k.groupTables.Range(func(_, v any) bool {
+		tables := v.(*sync.Map)
+		tables.Range(func(key, value any) bool {
+			lb := value.(lockTable)
+			bind := lb.getBind()
+			if bind.ServiceID != k.serviceID {
+				services[bind.ServiceID] = bind
+			}
+			return true
+		})
 		return true
 	})
 	if len(services) == 0 {
@@ -164,12 +168,16 @@ func (k *lockTableKeeper) doKeepRemoteLock(
 
 func (k *lockTableKeeper) doKeepLockTableBind(ctx context.Context) {
 	if !k.canDoKeep {
-		k.tables.Range(func(key, value any) bool {
-			lb := value.(lockTable)
-			bind := lb.getBind()
-			if bind.ServiceID == k.serviceID {
-				k.canDoKeep = true
-			}
+		k.groupTables.Range(func(_, v any) bool {
+			tables := v.(*sync.Map)
+			tables.Range(func(key, value any) bool {
+				lb := value.(lockTable)
+				bind := lb.getBind()
+				if bind.ServiceID == k.serviceID {
+					k.canDoKeep = true
+				}
+				return true
+			})
 			return true
 		})
 	}
@@ -197,14 +205,18 @@ func (k *lockTableKeeper) doKeepLockTableBind(ctx context.Context) {
 	}
 
 	n := 0
-	k.tables.Range(func(key, value any) bool {
-		lb := value.(lockTable)
-		bind := lb.getBind()
-		if bind.ServiceID == k.serviceID {
-			k.tables.Delete(key)
-			lb.close()
-		}
-		n++
+	k.groupTables.Range(func(_, v any) bool {
+		tables := v.(*sync.Map)
+		tables.Range(func(key, value any) bool {
+			lb := value.(lockTable)
+			bind := lb.getBind()
+			if bind.ServiceID == k.serviceID {
+				tables.Delete(key)
+				lb.close()
+			}
+			n++
+			return true
+		})
 		return true
 	})
 	if n > 0 {
diff --git a/pkg/lockservice/lock_table_local.go b/pkg/lockservice/lock_table_local.go
index 5780ea3e6..ad5d71b51 100644
--- a/pkg/lockservice/lock_table_local.go
+++ b/pkg/lockservice/lock_table_local.go
@@ -37,11 +37,12 @@ const (
 
 // a localLockTable instance manages the locks on a table
 type localLockTable struct {
-	bind   pb.LockTable
-	fsp    *fixedSlicePool
-	clock  clock.Clock
-	events *waiterEvents
-	mu     struct {
+	bind      pb.LockTable
+	fsp       *fixedSlicePool
+	clock     clock.Clock
+	events    *waiterEvents
+	txnHolder activeTxnHolder
+	mu        struct {
 		sync.RWMutex
 		closed           bool
 		store            LockStorage
@@ -53,12 +54,14 @@ func newLocalLockTable(
 	bind pb.LockTable,
 	fsp *fixedSlicePool,
 	events *waiterEvents,
-	clock clock.Clock) lockTable {
+	clock clock.Clock,
+	txnHolder activeTxnHolder) lockTable {
 	l := &localLockTable{
-		bind:   bind,
-		fsp:    fsp,
-		clock:  clock,
-		events: events,
+		bind:      bind,
+		fsp:       fsp,
+		clock:     clock,
+		events:    events,
+		txnHolder: txnHolder,
 	}
 	l.mu.store = newBtreeBasedStorage()
 	l.mu.tableCommittedAt, _ = clock.Now()
@@ -174,12 +177,22 @@ func (l *localLockTable) doLock(
 func (l *localLockTable) unlock(
 	txn *activeTxn,
 	ls *cowSlice,
-	commitTS timestamp.Timestamp) {
+	commitTS timestamp.Timestamp,
+	mutations ...pb.ExtraMutation) {
 	start := time.Now()
 	defer func() {
 		v2.TxnUnlockBtreeTotalDurationHistogram.Observe(time.Since(start).Seconds())
 	}()
 
+	getMutation := func(key []byte) int {
+		for i := range mutations {
+			if bytes.Equal(mutations[i].Key, key) {
+				return i
+			}
+		}
+		return -1
+	}
+
 	logUnlockTableOnLocal(
 		l.bind.ServiceID,
 		txn,
@@ -196,7 +209,7 @@ func (l *localLockTable) unlock(
 		return
 	}
 
-	b, ok := txn.holdBinds[l.bind.Table]
+	b, ok := txn.getHoldLocksLocked(l.bind.Group).tableBinds[l.bind.Table]
 	if !ok {
 		panic("BUG: missing bind")
 	}
@@ -204,6 +217,11 @@ func (l *localLockTable) unlock(
 	var startKey []byte
 	locks.iter(func(key []byte) bool {
 		if lock, ok := l.mu.store.Get(key); ok {
+			idx := getMutation(key)
+			if idx != -1 && mutations[idx].Skip {
+				return true
+			}
+
 			if lock.isLockRangeStart() {
 				startKey = key
 				return true
@@ -231,10 +249,31 @@ func (l *localLockTable) unlock(
 				panic("BUG: missing range end key")
 			}
 
+			if idx != -1 {
+				replaceTo := mutations[idx].ReplaceTo
+				lock.holders.replace(txn.txnID,
+					pb.WaitTxn{TxnID: replaceTo, CreatedOn: txn.remoteService})
+				// cannot dead lock here, the replaceTo txn was created on the same cn.
+				replaceToTxn := l.txnHolder.getActiveTxn(mutations[idx].ReplaceTo, true, txn.remoteService)
+				replaceToTxn.Lock()
+				replaceToTxn.lockAdded(l.bind.Group, l.bind, [][]byte{key})
+				replaceToTxn.Unlock()
+				return true
+			}
+
 			lockCanRemoved := lock.closeTxn(
 				txn,
 				notifyValue{ts: commitTS})
 			logLockUnlocked(txn, key, lock)
+
+			if l.bind.OriginTable == 2 {
+				var buf bytes.Buffer
+				lock.waiters.iter(func(w *waiter) bool {
+					buf.WriteString(w.String())
+					buf.WriteString(" ")
+					return true
+				})
+			}
 			if lockCanRemoved {
 				v2.TxnHoldLockDurationHistogram.Observe(time.Since(lock.createAt).Seconds())
 				l.mu.store.Delete(key)
@@ -328,7 +367,7 @@ func (l *localLockTable) acquireRowLockLocked(c *lockContext) error {
 				// only new holder can added lock into txn.
 				// newHolder is false means prev op of txn has already added lock into txn
 				if newHolder {
-					c.txn.lockAdded(l.bind, [][]byte{key})
+					c.txn.lockAdded(l.bind.Group, l.bind, [][]byte{key})
 				}
 				continue
 			}
@@ -395,7 +434,7 @@ func (l *localLockTable) addRowLockLocked(
 
 	// we must first add the lock to txn to ensure that the
 	// lock can be read when the deadlock is detected.
-	c.txn.lockAdded(l.bind, [][]byte{row})
+	c.txn.lockAdded(l.bind.Group, l.bind, [][]byte{row})
 	l.mu.store.Add(row, lock)
 }
 
@@ -439,7 +478,7 @@ func (l *localLockTable) addRangeLockLocked(
 				c.w = nil
 			}
 			if newHolder {
-				c.txn.lockAdded(l.bind, [][]byte{start, end})
+				c.txn.lockAdded(l.bind.Group, l.bind, [][]byte{start, end})
 			}
 			return nil, Lock{}, nil
 		}
@@ -512,7 +551,7 @@ func (l *localLockTable) addRangeLockLocked(
 				// only new holder can added lock into txn.
 				// newHolder is false means prev op of txn has already added lock into txn
 				if newHolder {
-					c.txn.lockAdded(l.bind, [][]byte{conflictKey})
+					c.txn.lockAdded(l.bind.Group, l.bind, [][]byte{conflictKey})
 				}
 				conflictWith = Lock{}
 				conflictKey = nil
@@ -547,7 +586,7 @@ func (l *localLockTable) addRangeLockLocked(
 	endLock.waiters = wq
 
 	// similar to row lock
-	c.txn.lockAdded(l.bind, [][]byte{start, end})
+	c.txn.lockAdded(l.bind.Group, l.bind, [][]byte{start, end})
 
 	l.mu.store.Add(start, startLock)
 	l.mu.store.Add(end, endLock)
@@ -665,6 +704,7 @@ func (c *mergeContext) commit(
 	}
 	txn.lockRemoved(
 		bind.ServiceID,
+		bind.Group,
 		bind.Table,
 		c.mergedLocks)
 
diff --git a/pkg/lockservice/lock_table_proxy.go b/pkg/lockservice/lock_table_proxy.go
new file mode 100644
index 000000000..0e04024c5
--- /dev/null
+++ b/pkg/lockservice/lock_table_proxy.go
@@ -0,0 +1,286 @@
+// Copyright 2023 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lockservice
+
+import (
+	"bytes"
+	"context"
+	"sync"
+
+	"github.com/matrixorigin/matrixone/pkg/common/util"
+	pb "github.com/matrixorigin/matrixone/pkg/pb/lock"
+	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
+)
+
+type localLockTableProxy struct {
+	remote    lockTable
+	serviceID string
+
+	mu struct {
+		sync.RWMutex
+		holders       map[string]*sharedOps // key: row
+		currentHolder map[string][]byte
+	}
+}
+
+func newLockTableProxy(
+	serviceID string,
+	remote lockTable) lockTable {
+	lp := &localLockTableProxy{
+		remote:    remote,
+		serviceID: serviceID,
+	}
+	lp.mu.holders = make(map[string]*sharedOps)
+	lp.mu.currentHolder = make(map[string][]byte)
+	return lp
+}
+
+func (lp *localLockTableProxy) lock(
+	ctx context.Context,
+	txn *activeTxn,
+	rows [][]byte,
+	options LockOptions,
+	cb func(pb.Result, error)) {
+	if options.Mode != pb.LockMode_Shared {
+		lp.remote.lock(ctx, txn, rows, options, cb)
+		return
+	}
+
+	if len(rows) != 1 {
+		panic("local lock table proxy can only support on single row")
+	}
+
+	lp.mu.Lock()
+	key := util.UnsafeBytesToString(rows[0])
+	v, ok := lp.mu.holders[key]
+	if !ok {
+		v = &sharedOps{
+			rows: rows,
+			bind: lp.getBind(),
+		}
+		lp.mu.holders[key] = v
+	}
+
+	first := v.isEmpty()
+	r := v.result
+	w := v.add(
+		lp.serviceID,
+		txn,
+		cb,
+		lp.hasRemoteHolderLocked(key))
+	if w != nil {
+		defer w.close()
+	}
+	lp.mu.Unlock()
+
+	if first {
+		lp.remote.lock(
+			ctx,
+			txn,
+			rows,
+			options,
+			func(r pb.Result, e error) {
+				lp.mu.Lock()
+				defer lp.mu.Unlock()
+				if e == nil {
+					lp.mu.currentHolder[key] = v.txns[0].txnID
+				}
+				v.done(r, e)
+			})
+		return
+	}
+
+	defer func() {
+		bind := lp.getBind()
+		txn.lockAdded(bind.Group, bind, rows)
+	}()
+
+	// wait first done
+	if w != nil {
+		w.wait(ctx)
+		return
+	}
+
+	cb(r, nil)
+}
+
+func (lp *localLockTableProxy) unlock(
+	txn *activeTxn,
+	ls *cowSlice,
+	commitTS timestamp.Timestamp,
+	_ ...pb.ExtraMutation) {
+	rows := ls.slice()
+	defer rows.unref()
+
+	skipped := 0
+	n := rows.len()
+	var mutations []pb.ExtraMutation
+	lp.mu.Lock()
+	defer lp.mu.Unlock()
+	rows.iter(func(key []byte) bool {
+		row := util.UnsafeBytesToString(key)
+		if v, ok := lp.mu.holders[row]; ok {
+			isHolder := lp.isRemoteHolderLocked(row, txn.txnID)
+			if !v.remove(txn) {
+				return true
+			}
+
+			// not the holder, no need to unlock
+			if !isHolder {
+				skipped++
+				if n > 1 {
+					mutations = append(mutations,
+						pb.ExtraMutation{
+							Key:  key,
+							Skip: true,
+						})
+				}
+				return true
+			}
+
+			// update holder to last txn
+			if !v.isEmpty() {
+				lp.mu.currentHolder[row] = v.last()
+				mutations = append(mutations,
+					pb.ExtraMutation{
+						Key:       key,
+						Skip:      false,
+						ReplaceTo: v.last(),
+					})
+			} else {
+				delete(lp.mu.currentHolder, row)
+			}
+		}
+		return true
+	})
+
+	// all skipped
+	if skipped == rows.len() {
+		return
+	}
+
+	lp.remote.unlock(txn, ls, commitTS, mutations...)
+}
+
+func (lp *localLockTableProxy) isRemoteHolderLocked(
+	row string,
+	txnID []byte) bool {
+	v, ok := lp.mu.currentHolder[row]
+	if !ok {
+		return false
+	}
+	return bytes.Equal(txnID, v)
+}
+
+func (lp *localLockTableProxy) hasRemoteHolderLocked(row string) bool {
+	_, ok := lp.mu.currentHolder[row]
+	return ok
+}
+
+func (lp *localLockTableProxy) getLock(
+	key []byte,
+	txn pb.WaitTxn,
+	fn func(Lock)) {
+	lp.remote.getLock(key, txn, fn)
+}
+
+func (lp *localLockTableProxy) getBind() pb.LockTable {
+	return lp.remote.getBind()
+}
+
+func (lp *localLockTableProxy) close() {
+	lp.remote.close()
+}
+
+type sharedOps struct {
+	bind    pb.LockTable
+	result  pb.Result
+	rows    [][]byte
+	txns    []*activeTxn
+	waiters []*waiter
+	cbs     []func(pb.Result, error)
+}
+
+func (s *sharedOps) done(
+	r pb.Result,
+	err error) {
+	for idx, cb := range s.cbs {
+		cb(r, err)
+		if idx > 0 {
+			s.waiters[idx].notify(notifyValue{})
+		}
+		s.cbs[idx] = nil
+		s.waiters[idx] = nil
+	}
+	if err != nil {
+		s.txns = s.txns[:0]
+		s.cbs = s.cbs[:0]
+		s.waiters = s.waiters[:0]
+	} else {
+		s.result = r
+	}
+}
+
+func (s *sharedOps) isEmpty() bool {
+	return len(s.txns) == 0
+}
+
+func (s *sharedOps) last() []byte {
+	return s.txns[len(s.txns)-1].txnID
+}
+
+func (s *sharedOps) add(
+	serviceID string,
+	txn *activeTxn,
+	cb func(pb.Result, error),
+	hasHolder bool) *waiter {
+	var w *waiter
+	if !hasHolder && !s.isEmpty() {
+		v := txn.toWaitTxn(serviceID, true)
+		w = acquireWaiter(v)
+		w.setStatus(blocking)
+	}
+	if hasHolder {
+		cb = nil
+	}
+
+	s.txns = append(s.txns, txn)
+	s.cbs = append(s.cbs, cb)
+	s.waiters = append(s.waiters, w)
+	return w
+}
+
+func (s *sharedOps) remove(txn *activeTxn) bool {
+	found := false
+	newTxns := s.txns[:0]
+	newCbs := s.cbs[:0]
+	newWaiters := s.waiters[:0]
+	for idx, v := range s.txns {
+		if v != txn {
+			if bytes.Equal(v.txnID, txn.txnID) {
+				panic("fatal")
+			}
+			newTxns = append(newTxns, v)
+			newWaiters = append(newWaiters, s.waiters[idx])
+			newCbs = append(newCbs, s.cbs[idx])
+		} else {
+			found = true
+		}
+	}
+	s.txns = newTxns
+	s.waiters = newWaiters
+	s.cbs = newCbs
+	return found
+}
diff --git a/pkg/lockservice/lock_table_remote.go b/pkg/lockservice/lock_table_remote.go
index 417e0a195..5f9177d0d 100644
--- a/pkg/lockservice/lock_table_remote.go
+++ b/pkg/lockservice/lock_table_remote.go
@@ -105,14 +105,14 @@ func (l *remoteLockTable) lock(
 			return
 		}
 
-		txn.lockAdded(l.bind, rows)
+		txn.lockAdded(l.bind.Group, l.bind, rows)
 		logRemoteLockAdded(txn, rows, opts, l.bind)
 		cb(resp.Lock.Result, nil)
 		return
 	}
 
 	// encounter any error, we also added lock to txn, because we need unlock on remote
-	txn.lockAdded(l.bind, rows)
+	txn.lockAdded(l.bind.Group, l.bind, rows)
 
 	logRemoteLockFailed(txn, rows, opts, l.bind, err)
 	// encounter any error, we need try to check bind is valid.
@@ -125,14 +125,15 @@ func (l *remoteLockTable) lock(
 func (l *remoteLockTable) unlock(
 	txn *activeTxn,
 	ls *cowSlice,
-	commitTS timestamp.Timestamp) {
+	commitTS timestamp.Timestamp,
+	mutations ...pb.ExtraMutation) {
 	logUnlockTableOnRemote(
 		l.serviceID,
 		txn,
 		l.bind)
 	st := time.Now()
 	for {
-		err := l.doUnlock(txn, commitTS)
+		err := l.doUnlock(txn, commitTS, mutations...)
 		if err == nil {
 			return
 		}
@@ -186,7 +187,8 @@ func (l *remoteLockTable) getLock(
 
 func (l *remoteLockTable) doUnlock(
 	txn *activeTxn,
-	commitTS timestamp.Timestamp) error {
+	commitTS timestamp.Timestamp,
+	mutations ...pb.ExtraMutation) error {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultRPCTimeout)
 	defer cancel()
 
@@ -197,6 +199,7 @@ func (l *remoteLockTable) doUnlock(
 	req.LockTable = l.bind
 	req.Unlock.TxnID = txn.txnID
 	req.Unlock.CommitTS = commitTS
+	req.Unlock.Mutations = mutations
 
 	resp, err := l.client.Send(ctx, req)
 	if err == nil {
@@ -232,7 +235,9 @@ func (l *remoteLockTable) doGetLock(key []byte, txn pb.WaitTxn) (Lock, bool, err
 		}
 		lock.holders.add(txn)
 		for _, v := range resp.GetTxnLock.WaitingList {
-			lock.addWaiter(acquireWaiter(v))
+			w := acquireWaiter(v)
+			lock.addWaiter(w)
+			w.close()
 		}
 		return lock, true, nil
 	}
@@ -260,8 +265,11 @@ func (l *remoteLockTable) handleError(txnID []byte, err error) error {
 	// the bind is valid.
 	new, err := getLockTableBind(
 		l.client,
+		l.bind.Group,
 		l.bind.Table,
-		l.serviceID)
+		l.bind.OriginTable,
+		l.serviceID,
+		l.bind.Sharding)
 	if err != nil {
 		logGetRemoteBindFailed(l.bind.Table, err)
 		return oldError
diff --git a/pkg/lockservice/log.go b/pkg/lockservice/log.go
index cabddd843..4824c6549 100644
--- a/pkg/lockservice/log.go
+++ b/pkg/lockservice/log.go
@@ -343,14 +343,8 @@ func logTxnReadyToClose(
 	txn *activeTxn) {
 	logger := getWithSkipLogger()
 	if logger.Enabled(zap.DebugLevel) {
-		var tables []uint64
-		for t := range txn.holdLocks {
-			tables = append(tables, t)
-		}
-
 		logger.Debug("ready to unlock txn",
-			txnField(txn),
-			uint64ArrayField("tables", tables))
+			txnField(txn))
 	}
 }
 
@@ -548,19 +542,6 @@ func waitTxnArrayField(name string, values []pb.WaitTxn) zap.Field {
 	return zap.String(name, buffer.String())
 }
 
-func uint64ArrayField(name string, values []uint64) zap.Field {
-	var buffer bytes.Buffer
-	buffer.WriteString("[")
-	for idx, v := range values {
-		buffer.WriteString(fmt.Sprintf("%d", v))
-		if idx != len(values)-1 {
-			buffer.WriteString(",")
-		}
-	}
-	buffer.WriteString("]")
-	return zap.String(name, buffer.String())
-}
-
 func waiterArrayField(name string, values ...*waiter) zap.Field {
 	var buffer bytes.Buffer
 	buffer.WriteString("[")
diff --git a/pkg/lockservice/reuse.go b/pkg/lockservice/reuse.go
index 73754daa9..547665726 100644
--- a/pkg/lockservice/reuse.go
+++ b/pkg/lockservice/reuse.go
@@ -43,8 +43,7 @@ func init() {
 	reuse.CreatePool[activeTxn](
 		func() *activeTxn {
 			txn := &activeTxn{
-				holdLocks: make(map[uint64]*cowSlice),
-				holdBinds: make(map[uint64]pb.LockTable),
+				lockHolders: make(map[uint32]*tableLockHolder),
 			}
 			txn.RWMutex = &sync.RWMutex{}
 			return txn
diff --git a/pkg/lockservice/rpc.go b/pkg/lockservice/rpc.go
index a84d3cca4..af4ba1171 100644
--- a/pkg/lockservice/rpc.go
+++ b/pkg/lockservice/rpc.go
@@ -33,7 +33,7 @@ import (
 
 var (
 	defaultRPCTimeout    = time.Second * 10
-	defaultHandleWorkers = 4
+	defaultHandleWorkers = 16
 )
 
 func acquireRequest() *pb.Request {
@@ -332,6 +332,11 @@ func (s *server) onMessage(
 
 func (s *server) handle(ctx context.Context) {
 	fn := func(ctx requestCtx) {
+		start := time.Now()
+		defer func() {
+			v2.TxnLockWorkerHandleDurationHistogram.Observe(time.Since(start).Seconds())
+		}()
+
 		req := ctx.req
 		defer releaseRequest(req)
 		resp := getResponse(req)
@@ -376,7 +381,7 @@ func writeResponse(
 			zap.String("response", detail))
 	}
 	// after write, response will be released by rpc
-	if err := cs.Write(ctx, resp); err != nil {
+	if err := cs.AsyncWrite(resp); err != nil {
 		getLogger().Error("write response failed",
 			zap.Error(err),
 			zap.String("response", detail))
diff --git a/pkg/lockservice/service.go b/pkg/lockservice/service.go
index 7512d9608..91549fd36 100644
--- a/pkg/lockservice/service.go
+++ b/pkg/lockservice/service.go
@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"hash/crc64"
 	"sync"
 	"time"
 
@@ -25,7 +26,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	"github.com/matrixorigin/matrixone/pkg/common/util"
-	"github.com/matrixorigin/matrixone/pkg/defines"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/lock"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
 	"github.com/matrixorigin/matrixone/pkg/txn/clock"
@@ -34,21 +34,10 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/util/trace"
 )
 
-var (
-	// All shared tables is shared by all tenants. We will use 1+TableID(high 4 bytes) + tenantID(low 4 bytes)
-	// as table id to avoid conflict.
-	sharedTables = map[uint64]uint64{
-		// can not use catalog.MO_TABLES_ID, because cycle import.
-		1: 1<<63 | 1<<32,
-		2: 1<<63 | 2<<32,
-		3: 1<<63 | 3<<32,
-	}
-)
-
 type service struct {
 	cfg                  Config
 	serviceID            string
-	tables               sync.Map // table id -> locktable
+	tableGroups          sync.Map // group -> sync.Map -> table id -> locktable
 	activeTxnHolder      activeTxnHolder
 	fsp                  *fixedSlicePool
 	deadlockDetector     *detector
@@ -66,7 +55,7 @@ type service struct {
 
 	mu struct {
 		sync.RWMutex
-		allocating map[uint64]chan struct{}
+		allocating map[uint32]map[uint64]chan struct{}
 	}
 }
 
@@ -86,7 +75,7 @@ func NewLockService(cfg Config) LockService {
 			stopper.WithLogger(getLogger().RawLogger())),
 		fetchWhoWaitingListC: make(chan who, 10240),
 	}
-	s.mu.allocating = make(map[uint64]chan struct{})
+	s.mu.allocating = make(map[uint32]map[uint64]chan struct{})
 	s.activeTxnHolder = newMapBasedTxnHandler(s.serviceID, s.fsp)
 	s.deadlockDetector = newDeadlockDetector(
 		s.fetchTxnWaitingList,
@@ -108,6 +97,7 @@ func (s *service) Lock(
 	txnID []byte,
 	options pb.LockOptions) (pb.Result, error) {
 	v2.TxnLockTotalCounter.Inc()
+	options.Validate(rows)
 
 	start := time.Now()
 	defer func() {
@@ -118,13 +108,12 @@ func (s *service) Lock(
 	ctx, span := trace.Debug(ctx, "lockservice.lock")
 	defer span.End()
 
-	tableID = encodeSharedTableID(ctx, tableID)
 	if options.ForwardTo != "" {
 		return s.forwardLock(ctx, tableID, rows, txnID, options)
 	}
 
 	txn := s.activeTxnHolder.getActiveTxn(txnID, true, "")
-	l, err := s.getLockTableWithCreate(tableID, true)
+	l, err := s.getLockTableWithCreate(options.Group, tableID, rows, options.Sharding)
 	if err != nil {
 		return pb.Result{}, err
 	}
@@ -157,7 +146,8 @@ func (s *service) Lock(
 func (s *service) Unlock(
 	ctx context.Context,
 	txnID []byte,
-	commitTS timestamp.Timestamp) error {
+	commitTS timestamp.Timestamp,
+	mutations ...pb.ExtraMutation) error {
 	start := time.Now()
 	defer func() {
 		v2.TxnUnlockDurationHistogram.Observe(time.Since(start).Seconds())
@@ -178,7 +168,7 @@ func (s *service) Unlock(
 	}
 
 	defer logUnlockTxn(s.serviceID, txn)()
-	txn.close(s.serviceID, txnID, commitTS, s.getLockTable)
+	txn.close(s.serviceID, txnID, commitTS, s.getLockTable, mutations...)
 	// The deadlock detector will hold the deadlocked transaction that is aborted
 	// to avoid the situation where the deadlock detection is interfered with by
 	// the abort transaction. When a transaction is unlocked, the deadlock detector
@@ -199,10 +189,15 @@ func (s *service) Close() error {
 	var err error
 	s.stopOnce.Do(func() {
 		s.stopper.Stop()
-		s.tables.Range(func(key, value any) bool {
-			value.(lockTable).close()
+		s.tableGroups.Range(func(key, value any) bool {
+			tables := value.(*sync.Map)
+			tables.Range(func(key, value any) bool {
+				value.(lockTable).close()
+				return true
+			})
 			return true
 		})
+
 		if err = s.remote.client.Close(); err != nil {
 			return
 		}
@@ -266,52 +261,76 @@ func (s *service) abortDeadlockTxn(wait pb.WaitTxn, err error) {
 	activeTxn.abort(s.serviceID, wait, err)
 }
 
-func (s *service) getLockTable(tableID uint64) (lockTable, error) {
-	return s.getLockTableWithCreate(tableID, false)
+func (s *service) getLockTable(
+	group uint32,
+	tableID uint64) (lockTable, error) {
+	if v := s.loadLockTable(group, tableID); v != nil {
+		return v, nil
+	}
+	return s.waitLockTableBind(
+		group,
+		tableID,
+		false), nil
 }
 
 func (s *service) getAllocatingC(
+	group uint32,
 	tableID uint64,
 	locked bool) chan struct{} {
 	if !locked {
 		s.mu.RLock()
 		defer s.mu.RUnlock()
 	}
-	return s.mu.allocating[tableID]
+	if m, ok := s.mu.allocating[group]; ok {
+		return m[tableID]
+	}
+	return nil
 }
 
 func (s *service) waitLockTableBind(
+	group uint32,
 	tableID uint64,
 	locked bool) lockTable {
-	c := s.getAllocatingC(tableID, locked)
+	c := s.getAllocatingC(group, tableID, locked)
 	if c != nil {
 		<-c
 	}
-	return s.loadLockTable(tableID)
+	return s.loadLockTable(group, tableID)
 }
 
-func (s *service) getLockTableWithCreate(tableID uint64, create bool) (lockTable, error) {
-	if v := s.loadLockTable(tableID); v != nil {
-		return v, nil
+func (s *service) getLockTableWithCreate(
+	group uint32,
+	tableID uint64,
+	rows [][]byte,
+	sharding pb.Sharding) (lockTable, error) {
+	originTableID := tableID
+	if sharding == pb.Sharding_ByRow {
+		tableID = shardingByRow(rows[0])
 	}
-	if !create {
-		return s.waitLockTableBind(tableID, false), nil
+
+	if v := s.loadLockTable(group, tableID); v != nil {
+		return v, nil
 	}
 
 	var c chan struct{}
 	fn := func() lockTable {
 		s.mu.Lock()
-		waitC := s.getAllocatingC(tableID, true)
+		waitC := s.getAllocatingC(group, tableID, true)
 		if waitC != nil {
 			s.mu.Unlock()
 			<-waitC
-			return s.loadLockTable(tableID)
+			return s.loadLockTable(group, tableID)
 		}
 
-		v := s.loadLockTable(tableID)
+		v := s.loadLockTable(group, tableID)
 		if v == nil {
 			c = make(chan struct{})
-			s.mu.allocating[tableID] = c
+			m, ok := s.mu.allocating[group]
+			if !ok {
+				m = make(map[uint64]chan struct{})
+				s.mu.allocating[group] = m
+			}
+			m[tableID] = c
 		}
 		s.mu.Unlock()
 		return v
@@ -323,19 +342,21 @@ func (s *service) getLockTableWithCreate(tableID uint64, create bool) (lockTable
 	defer func() {
 		s.mu.Lock()
 		defer s.mu.Unlock()
-		delete(s.mu.allocating, tableID)
+		delete(s.mu.allocating[group], tableID)
 		close(c)
 	}()
 	bind, err := getLockTableBind(
 		s.remote.client,
+		group,
 		tableID,
-		s.serviceID)
+		originTableID,
+		s.serviceID,
+		sharding)
 	if err != nil {
 		return nil, err
 	}
-
 	l := s.createLockTableByBind(bind)
-	if _, loaded := s.tables.LoadOrStore(tableID, l); loaded {
+	if _, loaded := s.getTables(group).LoadOrStore(tableID, l); loaded {
 		getLogger().Fatal("BUG: cannot loaded lock table from tables")
 	}
 	return l, nil
@@ -344,7 +365,20 @@ func (s *service) getLockTableWithCreate(tableID uint64, create bool) (lockTable
 func (s *service) handleBindChanged(newBind pb.LockTable) {
 	var oldBind pb.LockTable
 	new := s.createLockTableByBind(newBind)
-	old, loaded := s.tables.Swap(newBind.Table, new)
+	var tables *sync.Map
+
+	v, ok := s.tableGroups.Load(newBind.Group)
+	if !ok {
+		tables = &sync.Map{}
+		v, loaded := tables.LoadOrStore(newBind.Table, new)
+		if loaded {
+			tables = v.(*sync.Map)
+		}
+	} else {
+		tables = v.(*sync.Map)
+	}
+
+	old, loaded := tables.Swap(newBind.Table, new)
 	if loaded {
 		old.(lockTable).close()
 		oldBind = old.(lockTable).getBind()
@@ -363,24 +397,45 @@ func (s *service) createLockTableByBind(bind pb.LockTable) lockTable {
 			bind,
 			s.fsp,
 			s.events,
-			s.clock)
+			s.clock,
+			s.activeTxnHolder)
 	} else {
-		return newRemoteLockTable(
+		remote := newRemoteLockTable(
 			s.serviceID,
 			s.cfg.RemoteLockTimeout.Duration,
 			bind,
 			s.remote.client,
 			s.handleBindChanged)
+		if !s.cfg.EnableRemoteLocalProxy {
+			return remote
+		}
+		return newLockTableProxy(s.serviceID, remote)
 	}
 }
 
-func (s *service) loadLockTable(tableID uint64) lockTable {
-	if v, ok := s.tables.Load(tableID); ok {
+func (s *service) loadLockTable(
+	group uint32,
+	tableID uint64) lockTable {
+	if v, ok := s.getTables(group).Load(tableID); ok {
 		return v.(lockTable)
 	}
 	return nil
 }
 
+func (s *service) getTables(group uint32) *sync.Map {
+	v, ok := s.tableGroups.Load(group)
+	if ok {
+		return v.(*sync.Map)
+	}
+
+	tables := &sync.Map{}
+	v, loaded := s.tableGroups.LoadOrStore(group, tables)
+	if loaded {
+		return v.(*sync.Map)
+	}
+	return tables
+}
+
 type activeTxnHolder interface {
 	close()
 	getActiveTxn(txnID []byte, create bool, remoteService string) *activeTxn
@@ -556,31 +611,6 @@ func getUUIDFromServiceIdentifier(id string) string {
 	return id[19:]
 }
 
-func encodeSharedTableID(
-	ctx context.Context,
-	tableID uint64) uint64 {
-	v, ok := sharedTables[tableID]
-	if !ok {
-		return tableID
-	}
-
-	tenantID, ok := ctx.Value(defines.TenantIDKey{}).(uint32)
-	if !ok {
-		tenantID = 0
-	}
-
-	return v | uint64(tenantID)
-}
-
-func decodeSharedTableID(tableID uint64) (tenantID uint32, sharedTableID uint64, ok bool) {
-	if tableID&(1<<63) == 0 {
-		return 0, 0, false
-	}
-	tableID = tableID << 1 >> 1
-	return uint32(tableID << 32 >> 32), tableID >> 32, true
-}
-
-func isSharedTable(id uint64) bool {
-	_, ok := sharedTables[id]
-	return ok
+func shardingByRow(row []byte) uint64 {
+	return crc64.Checksum(row, crc64.MakeTable(crc64.ECMA))
 }
diff --git a/pkg/lockservice/service_forward.go b/pkg/lockservice/service_forward.go
index 4c6def488..8c7503b06 100644
--- a/pkg/lockservice/service_forward.go
+++ b/pkg/lockservice/service_forward.go
@@ -26,7 +26,11 @@ func (s *service) forwardLock(
 	rows [][]byte,
 	txnID []byte,
 	opts pb.LockOptions) (pb.Result, error) {
-	l, err := s.getLockTableWithCreate(tableID, true)
+	l, err := s.getLockTableWithCreate(
+		opts.Group,
+		tableID,
+		rows,
+		opts.Sharding)
 	if err != nil {
 		return pb.Result{}, err
 	}
diff --git a/pkg/lockservice/service_observability.go b/pkg/lockservice/service_observability.go
index 302f5036d..360e20b65 100644
--- a/pkg/lockservice/service_observability.go
+++ b/pkg/lockservice/service_observability.go
@@ -16,6 +16,7 @@ package lockservice
 
 import (
 	"context"
+	"sync"
 
 	pb "github.com/matrixorigin/matrixone/pkg/pb/lock"
 )
@@ -62,18 +63,24 @@ func (s *service) ForceRefreshLockTableBinds(targets ...uint64) {
 		return false
 	}
 
-	s.tables.Range(func(key, value any) bool {
-		id := key.(uint64)
-		if contains(id) {
-			value.(lockTable).close()
-			s.tables.Delete(key)
-		}
+	s.tableGroups.Range(func(_, v any) bool {
+		m := v.(*sync.Map)
+		m.Range(func(key, value any) bool {
+			id := key.(uint64)
+			if contains(id) {
+				value.(lockTable).close()
+				m.Delete(key)
+			}
+			return true
+		})
 		return true
 	})
 }
 
-func (s *service) GetLockTableBind(tableID uint64) (pb.LockTable, error) {
-	l, err := s.getLockTable(tableID)
+func (s *service) GetLockTableBind(
+	group uint32,
+	tableID uint64) (pb.LockTable, error) {
+	l, err := s.getLockTable(group, tableID)
 	if err != nil {
 		return pb.LockTable{}, err
 	}
@@ -84,27 +91,30 @@ func (s *service) GetLockTableBind(tableID uint64) (pb.LockTable, error) {
 }
 
 func (s *service) IterLocks(fn func(tableID uint64, keys [][]byte, lock Lock) bool) {
-	s.tables.Range(func(key, value any) bool {
-		tableID := key.(uint64)
-		l, ok := value.(lockTable).(*localLockTable)
-		if !ok {
-			return true
-		}
-		keys := make([][]byte, 0, 2)
-		return func() bool {
-			stop := false
-			l.mu.Lock()
-			defer l.mu.Unlock()
-			l.mu.store.Iter(func(key []byte, lock Lock) bool {
-				keys = append(keys, key)
-				if lock.isLockRangeStart() {
-					return true
-				}
-				stop = !fn(tableID, keys, lock)
-				keys = keys[:0]
+	s.tableGroups.Range(func(_, v any) bool {
+		m := v.(*sync.Map)
+		m.Range(func(key, value any) bool {
+			l, ok := value.(lockTable).(*localLockTable)
+			if !ok {
+				return true
+			}
+			keys := make([][]byte, 0, 2)
+			return func() bool {
+				stop := false
+				l.mu.Lock()
+				defer l.mu.Unlock()
+				l.mu.store.Iter(func(key []byte, lock Lock) bool {
+					keys = append(keys, key)
+					if lock.isLockRangeStart() {
+						return true
+					}
+					stop = !fn(l.bind.OriginTable, keys, lock)
+					keys = keys[:0]
+					return !stop
+				})
 				return !stop
-			})
-			return !stop
-		}()
+			}()
+		})
+		return true
 	})
 }
diff --git a/pkg/lockservice/service_remote.go b/pkg/lockservice/service_remote.go
index 9dfc2632f..d71a3cc8a 100644
--- a/pkg/lockservice/service_remote.go
+++ b/pkg/lockservice/service_remote.go
@@ -58,7 +58,7 @@ func (s *service) initRemote() {
 		rpcClient,
 		s.cfg.KeepBindDuration.Duration,
 		s.cfg.KeepRemoteLockDuration.Duration,
-		&s.tables)
+		&s.tableGroups)
 	s.initRemoteHandler()
 	if err := s.remote.server.Start(); err != nil {
 		panic(err)
@@ -127,7 +127,9 @@ func (s *service) handleForwardLock(
 	req *pb.Request,
 	resp *pb.Response,
 	cs morpc.ClientSession) {
-	l, err := s.getLockTable(req.LockTable.Table)
+	l, err := s.getLockTable(
+		req.LockTable.Group,
+		req.LockTable.Table)
 	if err != nil ||
 		l == nil {
 		// means that the lockservice sending the lock request holds a stale
@@ -175,7 +177,7 @@ func (s *service) handleRemoteUnlock(
 		writeResponse(ctx, cancel, resp, err, cs)
 		return
 	}
-	err = s.Unlock(ctx, req.Unlock.TxnID, req.Unlock.CommitTS)
+	err = s.Unlock(ctx, req.Unlock.TxnID, req.Unlock.CommitTS, req.Unlock.Mutations...)
 	writeResponse(ctx, cancel, resp, err, cs)
 }
 
@@ -243,7 +245,9 @@ func (s *service) handleKeepRemoteLock(
 func (s *service) getLocalLockTable(
 	req *pb.Request,
 	resp *pb.Response) (lockTable, error) {
-	l, err := s.getLockTableWithCreate(req.LockTable.Table, false)
+	l, err := s.getLockTable(
+		req.LockTable.Group,
+		req.LockTable.Table)
 	if err != nil {
 		return nil, err
 	}
@@ -271,7 +275,7 @@ func (s *service) getLocalLockTable(
 		uuidRequest := getUUIDFromServiceIdentifier(bind.ServiceID)
 		if strings.EqualFold(uuid, uuidRequest) {
 			l.close()
-			s.tables.Delete(bind.Table)
+			s.getTables(bind.Group).Delete(bind.Table)
 			return nil, ErrLockTableBindChanged
 		}
 
@@ -341,8 +345,11 @@ func (s *service) unlockTimeoutRemoteTxn(ctx context.Context) {
 
 func getLockTableBind(
 	c Client,
+	group uint32,
 	tableID uint64,
-	serviceID string) (pb.LockTable, error) {
+	originTableID uint64,
+	serviceID string,
+	sharding pb.Sharding) (pb.LockTable, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultRPCTimeout)
 	defer cancel()
 
@@ -352,6 +359,9 @@ func getLockTableBind(
 	req.Method = pb.Method_GetBind
 	req.GetBind.ServiceID = serviceID
 	req.GetBind.Table = tableID
+	req.GetBind.OriginTable = originTableID
+	req.GetBind.Sharding = sharding
+	req.GetBind.Group = group
 
 	resp, err := c.Send(ctx, req)
 	if err != nil {
diff --git a/pkg/lockservice/txn.go b/pkg/lockservice/txn.go
index 23b8c50ee..c074c99a1 100644
--- a/pkg/lockservice/txn.go
+++ b/pkg/lockservice/txn.go
@@ -31,6 +31,11 @@ var (
 	parallelUnlockTables = 2
 )
 
+type tableLockHolder struct {
+	tableKeys  map[uint64]*cowSlice
+	tableBinds map[uint64]pb.LockTable
+}
+
 // activeTxn one goroutine write, multi goroutine read
 type activeTxn struct {
 	*sync.RWMutex
@@ -38,8 +43,7 @@ type activeTxn struct {
 	txnKey         string
 	fsp            *fixedSlicePool
 	blockedWaiters []*waiter
-	holdLocks      map[uint64]*cowSlice
-	holdBinds      map[uint64]pb.LockTable
+	lockHolders    map[uint32]*tableLockHolder
 	remoteService  string
 	deadlockFound  bool
 }
@@ -65,9 +69,11 @@ func (txn activeTxn) Name() string {
 
 func (txn *activeTxn) lockRemoved(
 	serviceID string,
+	group uint32,
 	table uint64,
 	removedLocks map[string]struct{}) {
-	v, ok := txn.holdLocks[table]
+	h := txn.getHoldLocksLocked(group)
+	v, ok := h.tableKeys[table]
 	if !ok {
 		return
 	}
@@ -81,10 +87,11 @@ func (txn *activeTxn) lockRemoved(
 		return true
 	})
 	v.close()
-	txn.holdLocks[table] = newV
+	h.tableKeys[table] = newV
 }
 
 func (txn *activeTxn) lockAdded(
+	group uint32,
 	bind pb.LockTable,
 	locks [][]byte) {
 
@@ -105,65 +112,69 @@ func (txn *activeTxn) lockAdded(
 	//    the lock information. We use mutex to solve it.
 
 	defer logTxnLockAdded(txn, locks)
-	v, ok := txn.holdLocks[bind.Table]
+	h := txn.getHoldLocksLocked(group)
+	v, ok := h.tableKeys[bind.Table]
 	if ok {
 		v.append(locks)
 		return
 	}
-	txn.holdLocks[bind.Table] = newCowSlice(txn.fsp, locks)
-	txn.holdBinds[bind.Table] = bind
+	h.tableKeys[bind.Table] = newCowSlice(txn.fsp, locks)
+	h.tableBinds[bind.Table] = bind
 }
 
 func (txn *activeTxn) close(
 	serviceID string,
 	txnID []byte,
 	commitTS timestamp.Timestamp,
-	lockTableFunc func(uint64) (lockTable, error)) error {
+	lockTableFunc func(uint32, uint64) (lockTable, error),
+	mutations ...pb.ExtraMutation) error {
 	logTxnReadyToClose(serviceID, txn)
 
 	// cancel all blocked waiters
 	txn.cancelBlocks()
 
-	n := len(txn.holdLocks)
+	n := len(txn.lockHolders)
 	var wg sync.WaitGroup
 	v2.TxnUnlockTableTotalHistogram.Observe(float64(n))
-	for table, cs := range txn.holdLocks {
-		l, err := lockTableFunc(table)
-		if err != nil {
-			// if a remote transaction, then the corresponding locktable should be local
-			// and cannot return an error.
-			//
-			// or a local transaction holds a lock on remote lock table, but can not get the remote
-			// LockTable, it is a bug.
-			panic(err)
-		}
-		if l == nil {
-			continue
-		}
+	for group, h := range txn.lockHolders {
+		for table, cs := range h.tableKeys {
+			l, err := lockTableFunc(group, table)
+			if err != nil {
+				// if a remote transaction, then the corresponding locktable should be local
+				// and cannot return an error.
+				//
+				// or a local transaction holds a lock on remote lock table, but can not get the remote
+				// LockTable, it is a bug.
+				panic(err)
+			}
+			if l == nil {
+				continue
+			}
 
-		fn := func(table uint64, cs *cowSlice) func() {
-			return func() {
-				logTxnUnlockTable(
-					serviceID,
-					txn,
-					table)
-				l.unlock(txn, cs, commitTS)
-				logTxnUnlockTableCompleted(
-					serviceID,
-					txn,
-					table,
-					cs)
-				if n > parallelUnlockTables {
-					wg.Done()
+			fn := func(table uint64, cs *cowSlice, l lockTable) func() {
+				return func() {
+					logTxnUnlockTable(
+						serviceID,
+						txn,
+						table)
+					l.unlock(txn, cs, commitTS, mutations...)
+					logTxnUnlockTableCompleted(
+						serviceID,
+						txn,
+						table,
+						cs)
+					if n > parallelUnlockTables {
+						wg.Done()
+					}
 				}
 			}
-		}
 
-		if n > parallelUnlockTables {
-			wg.Add(1)
-			ants.Submit(fn(table, cs))
-		} else {
-			fn(table, cs)()
+			if n > parallelUnlockTables {
+				wg.Add(1)
+				ants.Submit(fn(table, cs, l))
+			} else {
+				fn(table, cs, l)()
+			}
 		}
 	}
 
@@ -176,11 +187,15 @@ func (txn *activeTxn) close(
 }
 
 func (txn *activeTxn) reset() {
-	for table, cs := range txn.holdLocks {
-		cs.close()
-		delete(txn.holdLocks, table)
-		delete(txn.holdBinds, table)
+	for g, h := range txn.lockHolders {
+		for table, cs := range h.tableKeys {
+			cs.close()
+			delete(h.tableKeys, table)
+			delete(h.tableBinds, table)
+		}
+		delete(txn.lockHolders, g)
 	}
+
 	txn.txnID = nil
 	txn.txnKey = ""
 	txn.blockedWaiters = txn.blockedWaiters[:0]
@@ -256,7 +271,7 @@ func (txn *activeTxn) fetchWhoWaitingMe(
 	txnID []byte,
 	holder activeTxnHolder,
 	waiters func(pb.WaitTxn) bool,
-	lockTableFunc func(uint64) (lockTable, error)) bool {
+	lockTableFunc func(uint32, uint64) (lockTable, error)) bool {
 	txn.RLock()
 	// txn already closed
 	if !bytes.Equal(txn.txnID, txnID) {
@@ -269,12 +284,18 @@ func (txn *activeTxn) fetchWhoWaitingMe(
 		txn.RUnlock()
 		panic("can not fetch waiting txn on remote txn")
 	}
-	tables := make([]uint64, 0, len(txn.holdLocks))
-	lockKeys := make([]*fixedSlice, 0, len(txn.holdLocks))
-	for table, cs := range txn.holdLocks {
-		tables = append(tables, table)
-		lockKeys = append(lockKeys, cs.slice())
+
+	groups := make([]uint32, 0, len(txn.lockHolders))
+	tables := make([]uint64, 0, len(txn.lockHolders))
+	lockKeys := make([]*fixedSlice, 0, len(txn.lockHolders))
+	for g, m := range txn.lockHolders {
+		for table, cs := range m.tableKeys {
+			tables = append(tables, table)
+			lockKeys = append(lockKeys, cs.slice())
+			groups = append(groups, g)
+		}
 	}
+
 	wt := txn.toWaitTxn(serviceID, true)
 	txn.RUnlock()
 
@@ -285,7 +306,7 @@ func (txn *activeTxn) fetchWhoWaitingMe(
 	}()
 
 	for idx, table := range tables {
-		l, err := lockTableFunc(table)
+		l, err := lockTableFunc(groups[idx], table)
 		if err != nil {
 			// if a remote transaction, then the corresponding locktable should be local
 			// and cannot return an error.
@@ -338,3 +359,16 @@ func (txn *activeTxn) getID() []byte {
 	defer txn.RUnlock()
 	return txn.txnID
 }
+
+func (txn *activeTxn) getHoldLocksLocked(group uint32) *tableLockHolder {
+	h, ok := txn.lockHolders[group]
+	if ok {
+		return h
+	}
+	h = &tableLockHolder{
+		tableKeys:  make(map[uint64]*cowSlice),
+		tableBinds: make(map[uint64]pb.LockTable),
+	}
+	txn.lockHolders[group] = h
+	return h
+}
diff --git a/pkg/lockservice/types.go b/pkg/lockservice/types.go
index eb2399725..d0f9f6826 100644
--- a/pkg/lockservice/types.go
+++ b/pkg/lockservice/types.go
@@ -100,7 +100,7 @@ type LockService interface {
 	Lock(ctx context.Context, tableID uint64, rows [][]byte, txnID []byte, options pb.LockOptions) (pb.Result, error)
 	// Unlock release all locks associated with the transaction. If commitTS is not empty, means
 	// the txn was committed.
-	Unlock(ctx context.Context, txnID []byte, commitTS timestamp.Timestamp) error
+	Unlock(ctx context.Context, txnID []byte, commitTS timestamp.Timestamp, mutations ...pb.ExtraMutation) error
 
 	// Close close the lock service.
 	Close() error
@@ -112,7 +112,7 @@ type LockService interface {
 	// ForceRefreshLockTableBinds force refresh all lock tables binds
 	ForceRefreshLockTableBinds(targets ...uint64)
 	// GetLockTableBind returns lock table bind
-	GetLockTableBind(tableID uint64) (pb.LockTable, error)
+	GetLockTableBind(group uint32, tableID uint64) (pb.LockTable, error)
 	// IterLocks iter all locks on current lock service. len(keys) == 2 if is range lock,
 	// len(keys) == 1 if is row lock. And keys only valid in current iter func call.
 	IterLocks(func(tableID uint64, keys [][]byte, lock Lock) bool)
@@ -137,7 +137,7 @@ type lockTable interface {
 	// 3. Other known errors.
 	lock(ctx context.Context, txn *activeTxn, rows [][]byte, options LockOptions, cb func(pb.Result, error))
 	// Unlock release a set of locks, if txn was committed, commitTS is not empty
-	unlock(txn *activeTxn, ls *cowSlice, commitTS timestamp.Timestamp)
+	unlock(txn *activeTxn, ls *cowSlice, commitTS timestamp.Timestamp, mutations ...pb.ExtraMutation)
 	// getLock get a lock
 	getLock(key []byte, txn pb.WaitTxn, fn func(Lock))
 	// getBind returns lock table binding
@@ -162,7 +162,7 @@ type lockTable interface {
 type LockTableAllocator interface {
 	// Get get the original LockTable data corresponding to a Table. If there is no
 	// corresponding binding, then the CN binding of the current request will be used.
-	Get(serviceID string, tableID uint64) pb.LockTable
+	Get(serviceID string, group uint32, tableID, originTableID uint64, sharding pb.Sharding) pb.LockTable
 	// KeepLockTableBind once a cn is bound to a Table, a heartbeat needs to be sent
 	// periodically to keep the binding in place. If no heartbeat is sent for a long
 	// period of time to maintain the binding, the binding will become invalid.
diff --git a/pkg/pb/lock/lock.go b/pkg/pb/lock/lock.go
index ec9010ce0..dd995e96f 100644
--- a/pkg/pb/lock/lock.go
+++ b/pkg/pb/lock/lock.go
@@ -262,3 +262,17 @@ func (m Request) Name() string {
 func (m Response) Name() string {
 	return "lockservice.response"
 }
+
+func (m LockOptions) Validate(rows [][]byte) {
+	if m.Sharding == Sharding_None {
+		return
+	}
+
+	if m.Granularity != Granularity_Row {
+		panic("cannot lock with sharding without row granularity")
+	}
+
+	if len(rows) != 1 {
+		panic("cannot lock with sharding without single row")
+	}
+}
diff --git a/pkg/pb/lock/lock.pb.go b/pkg/pb/lock/lock.pb.go
index f4011dd91..80923ff72 100644
--- a/pkg/pb/lock/lock.pb.go
+++ b/pkg/pb/lock/lock.pb.go
@@ -103,6 +103,31 @@ func (WaitPolicy) EnumDescriptor() ([]byte, []int) {
 	return fileDescriptor_164ad2988c7acaf1, []int{2}
 }
 
+type Sharding int32
+
+const (
+	Sharding_None  Sharding = 0
+	Sharding_ByRow Sharding = 1
+)
+
+var Sharding_name = map[int32]string{
+	0: "None",
+	1: "ByRow",
+}
+
+var Sharding_value = map[string]int32{
+	"None":  0,
+	"ByRow": 1,
+}
+
+func (x Sharding) String() string {
+	return proto.EnumName(Sharding_name, int32(x))
+}
+
+func (Sharding) EnumDescriptor() ([]byte, []int) {
+	return fileDescriptor_164ad2988c7acaf1, []int{3}
+}
+
 // Method lock table operations
 type Method int32
 
@@ -156,7 +181,7 @@ func (x Method) String() string {
 }
 
 func (Method) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_164ad2988c7acaf1, []int{3}
+	return fileDescriptor_164ad2988c7acaf1, []int{4}
 }
 
 // LockOptions lock options
@@ -166,6 +191,8 @@ type LockOptions struct {
 	Policy               WaitPolicy  `protobuf:"varint,3,opt,name=Policy,proto3,enum=lock.WaitPolicy" json:"Policy,omitempty"`
 	ForwardTo            string      `protobuf:"bytes,4,opt,name=ForwardTo,proto3" json:"ForwardTo,omitempty"`
 	TableDefChanged      bool        `protobuf:"varint,5,opt,name=TableDefChanged,proto3" json:"TableDefChanged,omitempty"`
+	Group                uint32      `protobuf:"varint,6,opt,name=Group,proto3" json:"Group,omitempty"`
+	Sharding             Sharding    `protobuf:"varint,7,opt,name=Sharding,proto3,enum=lock.Sharding" json:"Sharding,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}    `json:"-"`
 	XXX_unrecognized     []byte      `json:"-"`
 	XXX_sizecache        int32       `json:"-"`
@@ -239,6 +266,20 @@ func (m *LockOptions) GetTableDefChanged() bool {
 	return false
 }
 
+func (m *LockOptions) GetGroup() uint32 {
+	if m != nil {
+		return m.Group
+	}
+	return 0
+}
+
+func (m *LockOptions) GetSharding() Sharding {
+	if m != nil {
+		return m.Sharding
+	}
+	return Sharding_None
+}
+
 // LockTable describes which CN manages a Table's Locks.
 type LockTable struct {
 	// Table table id
@@ -249,6 +290,9 @@ type LockTable struct {
 	Version uint64 `protobuf:"varint,3,opt,name=Version,proto3" json:"Version,omitempty"`
 	// Valid false if the service is disabled, and no new service bind this table
 	Valid                bool     `protobuf:"varint,4,opt,name=Valid,proto3" json:"Valid,omitempty"`
+	Group                uint32   `protobuf:"varint,5,opt,name=Group,proto3" json:"Group,omitempty"`
+	Sharding             Sharding `protobuf:"varint,6,opt,name=Sharding,proto3,enum=lock.Sharding" json:"Sharding,omitempty"`
+	OriginTable          uint64   `protobuf:"varint,7,opt,name=OriginTable,proto3" json:"OriginTable,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -315,6 +359,27 @@ func (m *LockTable) GetValid() bool {
 	return false
 }
 
+func (m *LockTable) GetGroup() uint32 {
+	if m != nil {
+		return m.Group
+	}
+	return 0
+}
+
+func (m *LockTable) GetSharding() Sharding {
+	if m != nil {
+		return m.Sharding
+	}
+	return Sharding_None
+}
+
+func (m *LockTable) GetOriginTable() uint64 {
+	if m != nil {
+		return m.OriginTable
+	}
+	return 0
+}
+
 // Request is used to send a request for a LockTable related operation to another
 // service.
 type Request struct {
@@ -961,6 +1026,7 @@ type UnlockRequest struct {
 	// CommitTS is the commit timestamp of the current txn. Empty if txn is
 	// roll backed
 	CommitTS             timestamp.Timestamp `protobuf:"bytes,2,opt,name=CommitTS,proto3" json:"CommitTS"`
+	Mutations            []ExtraMutation     `protobuf:"bytes,3,rep,name=Mutations,proto3" json:"Mutations"`
 	XXX_NoUnkeyedLiteral struct{}            `json:"-"`
 	XXX_unrecognized     []byte              `json:"-"`
 	XXX_sizecache        int32               `json:"-"`
@@ -1013,6 +1079,13 @@ func (m *UnlockRequest) GetCommitTS() timestamp.Timestamp {
 	return timestamp.Timestamp{}
 }
 
+func (m *UnlockRequest) GetMutations() []ExtraMutation {
+	if m != nil {
+		return m.Mutations
+	}
+	return nil
+}
+
 // UnlockResponse unlock lock on remote lock service response. CN -> CN
 type UnlockResponse struct {
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -1057,6 +1130,9 @@ var xxx_messageInfo_UnlockResponse proto.InternalMessageInfo
 type GetBindRequest struct {
 	ServiceID            string   `protobuf:"bytes,1,opt,name=ServiceID,proto3" json:"ServiceID,omitempty"`
 	Table                uint64   `protobuf:"varint,2,opt,name=Table,proto3" json:"Table,omitempty"`
+	Group                uint32   `protobuf:"varint,3,opt,name=Group,proto3" json:"Group,omitempty"`
+	Sharding             Sharding `protobuf:"varint,4,opt,name=Sharding,proto3,enum=lock.Sharding" json:"Sharding,omitempty"`
+	OriginTable          uint64   `protobuf:"varint,5,opt,name=OriginTable,proto3" json:"OriginTable,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -1109,6 +1185,27 @@ func (m *GetBindRequest) GetTable() uint64 {
 	return 0
 }
 
+func (m *GetBindRequest) GetGroup() uint32 {
+	if m != nil {
+		return m.Group
+	}
+	return 0
+}
+
+func (m *GetBindRequest) GetSharding() Sharding {
+	if m != nil {
+		return m.Sharding
+	}
+	return Sharding_None
+}
+
+func (m *GetBindRequest) GetOriginTable() uint64 {
+	if m != nil {
+		return m.OriginTable
+	}
+	return 0
+}
+
 // GetBindResponse get bind request from allocator response. TN -> CN
 type GetBindResponse struct {
 	LockTable            LockTable `protobuf:"bytes,2,opt,name=LockTable,proto3" json:"LockTable"`
@@ -1437,10 +1534,74 @@ func (m *Result) GetTableDefChanged() bool {
 	return false
 }
 
+type ExtraMutation struct {
+	Key                  []byte   `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
+	Skip                 bool     `protobuf:"varint,2,opt,name=skip,proto3" json:"skip,omitempty"`
+	ReplaceTo            []byte   `protobuf:"bytes,3,opt,name=replaceTo,proto3" json:"replaceTo,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *ExtraMutation) Reset()         { *m = ExtraMutation{} }
+func (m *ExtraMutation) String() string { return proto.CompactTextString(m) }
+func (*ExtraMutation) ProtoMessage()    {}
+func (*ExtraMutation) Descriptor() ([]byte, []int) {
+	return fileDescriptor_164ad2988c7acaf1, []int{20}
+}
+func (m *ExtraMutation) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *ExtraMutation) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_ExtraMutation.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *ExtraMutation) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ExtraMutation.Merge(m, src)
+}
+func (m *ExtraMutation) XXX_Size() int {
+	return m.Size()
+}
+func (m *ExtraMutation) XXX_DiscardUnknown() {
+	xxx_messageInfo_ExtraMutation.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ExtraMutation proto.InternalMessageInfo
+
+func (m *ExtraMutation) GetKey() []byte {
+	if m != nil {
+		return m.Key
+	}
+	return nil
+}
+
+func (m *ExtraMutation) GetSkip() bool {
+	if m != nil {
+		return m.Skip
+	}
+	return false
+}
+
+func (m *ExtraMutation) GetReplaceTo() []byte {
+	if m != nil {
+		return m.ReplaceTo
+	}
+	return nil
+}
+
 func init() {
 	proto.RegisterEnum("lock.Granularity", Granularity_name, Granularity_value)
 	proto.RegisterEnum("lock.LockMode", LockMode_name, LockMode_value)
 	proto.RegisterEnum("lock.WaitPolicy", WaitPolicy_name, WaitPolicy_value)
+	proto.RegisterEnum("lock.Sharding", Sharding_name, Sharding_value)
 	proto.RegisterEnum("lock.Method", Method_name, Method_value)
 	proto.RegisterType((*LockOptions)(nil), "lock.LockOptions")
 	proto.RegisterType((*LockTable)(nil), "lock.LockTable")
@@ -1462,85 +1623,95 @@ func init() {
 	proto.RegisterType((*KeepRemoteLockRequest)(nil), "lock.KeepRemoteLockRequest")
 	proto.RegisterType((*KeepRemoteLockResponse)(nil), "lock.KeepRemoteLockResponse")
 	proto.RegisterType((*Result)(nil), "lock.Result")
+	proto.RegisterType((*ExtraMutation)(nil), "lock.ExtraMutation")
 }
 
 func init() { proto.RegisterFile("lock.proto", fileDescriptor_164ad2988c7acaf1) }
 
 var fileDescriptor_164ad2988c7acaf1 = []byte{
-	// 1158 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x57, 0xdd, 0x4e, 0x1b, 0x47,
-	0x14, 0x66, 0xed, 0xb5, 0xbd, 0x3e, 0x6b, 0xcc, 0x32, 0x85, 0x74, 0x9b, 0x46, 0xc4, 0x5d, 0x25,
-	0x92, 0x4b, 0x5a, 0x10, 0x50, 0xaa, 0xa8, 0x55, 0x52, 0x09, 0x08, 0x84, 0x90, 0x94, 0x68, 0x70,
-	0x53, 0xa9, 0x52, 0x2f, 0x16, 0x7b, 0x30, 0x2b, 0xec, 0x1d, 0x77, 0x77, 0x0d, 0xce, 0x1b, 0xf4,
-	0x29, 0xfa, 0x2e, 0xbd, 0xcb, 0x65, 0x6e, 0x2a, 0xf5, 0xaa, 0x6a, 0xb9, 0xe9, 0x6b, 0x54, 0xf3,
-	0xb3, 0x3f, 0xb3, 0x5e, 0x43, 0xd4, 0xbb, 0x39, 0xdf, 0xf9, 0x99, 0x73, 0x66, 0xbe, 0x3d, 0x67,
-	0x16, 0x60, 0x40, 0xbb, 0x17, 0x6b, 0xa3, 0x80, 0x46, 0x14, 0xe9, 0x6c, 0x7d, 0xf7, 0xcb, 0xbe,
-	0x17, 0x9d, 0x8f, 0x4f, 0xd7, 0xba, 0x74, 0xb8, 0xde, 0xa7, 0x7d, 0xba, 0xce, 0x95, 0xa7, 0xe3,
-	0x33, 0x2e, 0x71, 0x81, 0xaf, 0x84, 0xd3, 0xdd, 0x85, 0xc8, 0x1b, 0x92, 0x30, 0x72, 0x87, 0x23,
-	0x01, 0x38, 0x7f, 0x6a, 0x60, 0xbe, 0xa4, 0xdd, 0x8b, 0xe3, 0x51, 0xe4, 0x51, 0x3f, 0x44, 0x5b,
-	0x60, 0x1e, 0x04, 0xae, 0x3f, 0x1e, 0xb8, 0x81, 0x17, 0xbd, 0xb5, 0xb5, 0x96, 0xd6, 0x6e, 0x6e,
-	0x2e, 0xae, 0xf1, 0x7d, 0x33, 0x0a, 0x9c, 0xb5, 0x42, 0x0e, 0xe8, 0xaf, 0x68, 0x8f, 0xd8, 0x25,
-	0x6e, 0xdd, 0x14, 0xd6, 0x2c, 0x2a, 0x43, 0x31, 0xd7, 0xa1, 0x36, 0x54, 0x5f, 0xd3, 0x81, 0xd7,
-	0x7d, 0x6b, 0x97, 0xb9, 0x95, 0x25, 0xac, 0x7e, 0x74, 0xbd, 0x48, 0xe0, 0x58, 0xea, 0xd1, 0x3d,
-	0xa8, 0xef, 0xd3, 0xe0, 0xca, 0x0d, 0x7a, 0x1d, 0x6a, 0xeb, 0x2d, 0xad, 0x5d, 0xc7, 0x29, 0x80,
-	0xda, 0xb0, 0xd0, 0x71, 0x4f, 0x07, 0x64, 0x8f, 0x9c, 0xed, 0x9e, 0xbb, 0x7e, 0x9f, 0xf4, 0xec,
-	0x4a, 0x4b, 0x6b, 0x1b, 0x38, 0x0f, 0x3b, 0x14, 0xea, 0x2c, 0x07, 0x0e, 0xa3, 0x25, 0xa8, 0xf0,
-	0x05, 0xaf, 0x48, 0xc7, 0x42, 0x60, 0x5b, 0x9d, 0x90, 0xe0, 0xd2, 0xeb, 0x92, 0xc3, 0x3d, 0x9e,
-	0x7d, 0x1d, 0xa7, 0x00, 0xb2, 0xa1, 0xf6, 0x86, 0x04, 0xa1, 0x47, 0x7d, 0x9e, 0xb3, 0x8e, 0x63,
-	0x91, 0x45, 0x7b, 0xe3, 0x0e, 0xbc, 0x1e, 0x4f, 0xcf, 0xc0, 0x42, 0x70, 0x7e, 0xd7, 0xa1, 0x86,
-	0xc9, 0x2f, 0x63, 0x12, 0x46, 0x2c, 0xb2, 0x5c, 0x1e, 0xee, 0xc9, 0x3d, 0x53, 0x00, 0x6d, 0x65,
-	0x52, 0xe3, 0xfb, 0x9a, 0x9b, 0x0b, 0xe9, 0xa9, 0x71, 0x78, 0x47, 0x7f, 0xf7, 0xd7, 0xfd, 0x39,
-	0x9c, 0x29, 0xe1, 0x01, 0x54, 0x5f, 0x91, 0xe8, 0x9c, 0xf6, 0xe4, 0x09, 0x36, 0x84, 0x87, 0xc0,
-	0xb0, 0xd4, 0xa1, 0x47, 0xa0, 0x33, 0x17, 0x9e, 0x99, 0x19, 0xdf, 0x1c, 0x43, 0xe4, 0xee, 0x32,
-	0x2e, 0x37, 0x42, 0x1b, 0x50, 0xfd, 0xc1, 0x67, 0x16, 0xfc, 0x0c, 0xcd, 0xcd, 0x8f, 0x84, 0xb9,
-	0xc0, 0x54, 0x07, 0x69, 0x88, 0x9e, 0x00, 0x1c, 0x90, 0xa8, 0x33, 0xf1, 0xf9, 0x2e, 0x55, 0xee,
-	0xf6, 0xb1, 0xe4, 0x47, 0x82, 0xab, 0xae, 0x19, 0x07, 0x74, 0x08, 0xcd, 0x03, 0x12, 0xb1, 0x5b,
-	0xf7, 0xfc, 0xfe, 0x4b, 0x2f, 0x8c, 0xec, 0x1a, 0x0f, 0xf1, 0x69, 0x12, 0x22, 0xa3, 0x53, 0xc3,
-	0xe4, 0x1c, 0xd1, 0x57, 0x50, 0x3b, 0x20, 0xd1, 0x8e, 0xe7, 0xf7, 0x6c, 0x83, 0xc7, 0x58, 0x4a,
-	0x62, 0x30, 0x50, 0x75, 0x8e, 0x4d, 0x11, 0x86, 0xc5, 0x23, 0x42, 0x46, 0xe9, 0x39, 0x33, 0xff,
-	0x3a, 0xf7, 0x5f, 0x11, 0xfe, 0x53, 0x6a, 0x35, 0xd2, 0xb4, 0x3b, 0x2b, 0x8a, 0x81, 0x98, 0x0c,
-	0x69, 0x44, 0xf8, 0xb9, 0x40, 0xb6, 0x28, 0x55, 0x97, 0x2b, 0x4a, 0x55, 0x3a, 0x7f, 0xe8, 0x60,
-	0x60, 0x12, 0x8e, 0xa8, 0x1f, 0x92, 0x5b, 0x48, 0x94, 0xf2, 0xa1, 0x74, 0x03, 0x1f, 0x96, 0xa0,
-	0xf2, 0x2c, 0x08, 0x68, 0xc0, 0x49, 0xd3, 0xc0, 0x42, 0x40, 0x9f, 0x43, 0xed, 0x7b, 0x72, 0xc5,
-	0x6b, 0xd7, 0x0b, 0xe9, 0x87, 0x63, 0x3d, 0xfa, 0x42, 0x12, 0x4a, 0x30, 0x04, 0x65, 0x09, 0x25,
-	0xd2, 0x54, 0x18, 0xb5, 0x99, 0x30, 0xaa, 0x9a, 0xbd, 0x93, 0x98, 0x51, 0x8a, 0x47, 0x4c, 0xa9,
-	0xa7, 0x0a, 0xa5, 0x04, 0x1f, 0xec, 0x69, 0x4a, 0x29, 0xbe, 0x59, 0x4e, 0xbd, 0x98, 0xe2, 0x94,
-	0xe0, 0xc3, 0xbd, 0x62, 0x4e, 0x29, 0x71, 0xf2, 0xa4, 0xda, 0x4e, 0x49, 0x25, 0x48, 0xb1, 0x9c,
-	0x23, 0x95, 0xe2, 0x9d, 0xb0, 0xea, 0xa4, 0x88, 0x55, 0x82, 0x04, 0xf7, 0x67, 0xb2, 0x4a, 0x09,
-	0x55, 0x40, 0xab, 0x17, 0x53, 0xb4, 0x32, 0xb3, 0x75, 0xe5, 0x69, 0xa5, 0xd6, 0x95, 0xe3, 0xd5,
-	0xaf, 0xb2, 0xcf, 0xc7, 0xfd, 0x89, 0xf5, 0xc3, 0x89, 0x2f, 0x69, 0xd5, 0xc0, 0x42, 0xb8, 0xa5,
-	0x1f, 0x22, 0xd0, 0x31, 0xbd, 0x0a, 0xed, 0x72, 0xab, 0xdc, 0x6e, 0x60, 0xbe, 0x46, 0x1b, 0x50,
-	0x93, 0xa3, 0x63, 0xba, 0xe3, 0x48, 0x45, 0x7c, 0x56, 0x52, 0x74, 0xbe, 0x81, 0x46, 0x36, 0x61,
-	0xb4, 0x0a, 0x55, 0x4c, 0xc2, 0xf1, 0x20, 0xe2, 0xb9, 0x98, 0x31, 0x8f, 0x05, 0x16, 0x53, 0x45,
-	0x48, 0xce, 0xb7, 0xb0, 0x38, 0xd5, 0x65, 0x66, 0xd4, 0x62, 0x41, 0x19, 0xd3, 0x2b, 0x5e, 0x45,
-	0x03, 0xb3, 0xa5, 0xe3, 0x02, 0x9a, 0xe6, 0x93, 0xec, 0xe5, 0x63, 0x31, 0x19, 0x2a, 0x58, 0x08,
-	0x68, 0x1b, 0xcc, 0x2c, 0xa1, 0x4a, 0xad, 0x72, 0xdb, 0xdc, 0x9c, 0x4f, 0x67, 0x56, 0x67, 0xe2,
-	0xcb, 0xd4, 0xb2, 0x76, 0xce, 0x53, 0x58, 0x2e, 0x6c, 0x61, 0xe8, 0x21, 0x94, 0x3b, 0x13, 0x5f,
-	0x56, 0x58, 0x18, 0x87, 0xe9, 0x9d, 0x63, 0xb8, 0x53, 0x4c, 0xd7, 0x7c, 0x42, 0xda, 0x07, 0x26,
-	0xf4, 0x04, 0x6a, 0x52, 0x3b, 0xfb, 0xca, 0x77, 0x03, 0xe2, 0x46, 0xa4, 0x77, 0xec, 0xc7, 0x57,
-	0x9e, 0x00, 0xce, 0xcf, 0x30, 0xaf, 0x0c, 0x83, 0x19, 0x41, 0xbe, 0x06, 0x63, 0x97, 0x0e, 0x87,
-	0x5e, 0xd4, 0x39, 0x91, 0xe3, 0x6c, 0x69, 0x2d, 0x7d, 0x69, 0x74, 0xe2, 0x95, 0x4c, 0x30, 0xb1,
-	0x75, 0x2c, 0x68, 0xaa, 0x9d, 0xc1, 0xd9, 0xe3, 0xdf, 0x72, 0xa6, 0xeb, 0xaa, 0x9c, 0xd4, 0xf2,
-	0x9c, 0x4c, 0xe6, 0x7a, 0x29, 0x33, 0xd7, 0x9d, 0x7d, 0x58, 0xc8, 0x7d, 0xb0, 0xff, 0x6b, 0xe4,
-	0x3a, 0x8f, 0xc1, 0x9e, 0x35, 0x0d, 0x6e, 0xce, 0xcb, 0x79, 0x04, 0x9f, 0xcc, 0xfc, 0xe2, 0x51,
-	0x13, 0x4a, 0xc7, 0x47, 0xdc, 0xc7, 0xc0, 0xa5, 0xe3, 0x23, 0x67, 0x1b, 0x96, 0x0b, 0x67, 0xc4,
-	0x2d, 0x7b, 0xb4, 0xe1, 0x4e, 0x71, 0x0f, 0x98, 0xda, 0xe0, 0x5f, 0x2d, 0xfe, 0xc6, 0xd0, 0x06,
-	0x18, 0xcc, 0x94, 0x5f, 0xb7, 0x76, 0xd3, 0x31, 0x24, 0x66, 0xa8, 0x05, 0xe6, 0x73, 0x37, 0xdc,
-	0xa5, 0xfe, 0xd9, 0xc0, 0xeb, 0x46, 0xfc, 0xf0, 0x0c, 0x9c, 0x85, 0xd0, 0x03, 0x98, 0x7f, 0xee,
-	0x86, 0xaf, 0x03, 0x72, 0x29, 0xae, 0x96, 0x0f, 0x1b, 0x03, 0xab, 0x20, 0x7a, 0x0c, 0xf5, 0x84,
-	0x0a, 0xb2, 0x5b, 0xdc, 0x44, 0x93, 0xd4, 0xf8, 0xc3, 0x1f, 0x7d, 0xab, 0x9f, 0x29, 0xef, 0x57,
-	0x54, 0xe3, 0x4d, 0xc0, 0x9a, 0x43, 0x75, 0xa8, 0x60, 0x66, 0x61, 0x69, 0xab, 0x0f, 0xc5, 0x09,
-	0xf0, 0x57, 0xe9, 0x3c, 0xd4, 0x9f, 0x4d, 0xba, 0x83, 0x71, 0xe8, 0x5d, 0x12, 0x6b, 0x0e, 0x01,
-	0x54, 0x4f, 0xce, 0xdd, 0x80, 0xf4, 0x2c, 0x6d, 0xf5, 0x01, 0x40, 0xfa, 0x38, 0x45, 0x06, 0xe8,
-	0x4c, 0xb2, 0xe6, 0x50, 0x03, 0x8c, 0x7d, 0x37, 0x8c, 0xf6, 0x5d, 0x6f, 0x60, 0x69, 0xab, 0xbf,
-	0x69, 0xf1, 0x14, 0x66, 0x26, 0x2c, 0xae, 0x08, 0x23, 0x68, 0x6d, 0x69, 0xa8, 0x99, 0x1d, 0x6e,
-	0x56, 0x09, 0xa1, 0xfc, 0xb0, 0xb2, 0xca, 0x0c, 0x53, 0x2f, 0xd2, 0xd2, 0x91, 0x99, 0x0c, 0x22,
-	0xab, 0x82, 0x96, 0x0b, 0xc6, 0x8b, 0x55, 0x45, 0x0b, 0x60, 0xca, 0x87, 0x31, 0x77, 0xaa, 0xa1,
-	0x45, 0x98, 0x97, 0x80, 0xdc, 0xdf, 0xd8, 0xf9, 0xee, 0xfd, 0x3f, 0x2b, 0xda, 0xbb, 0xeb, 0x15,
-	0xed, 0xfd, 0xf5, 0x8a, 0xf6, 0xf7, 0xf5, 0x8a, 0xf6, 0x53, 0xf6, 0x97, 0x61, 0xe8, 0x46, 0x81,
-	0x37, 0xa1, 0x81, 0xd7, 0xf7, 0xfc, 0x58, 0xf0, 0xc9, 0xfa, 0xe8, 0xa2, 0xbf, 0x3e, 0x3a, 0x5d,
-	0x67, 0x21, 0x4e, 0xab, 0xfc, 0x47, 0x61, 0xeb, 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0x32, 0x55,
-	0xe6, 0xfa, 0x7c, 0x0c, 0x00, 0x00,
+	// 1300 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x57, 0xcb, 0x6e, 0x1b, 0x55,
+	0x18, 0xce, 0xd8, 0x63, 0x7b, 0xfc, 0x8f, 0xed, 0x4c, 0x0e, 0x49, 0x19, 0x4a, 0x95, 0x98, 0x51,
+	0x2a, 0x19, 0x17, 0x12, 0xc5, 0x21, 0x50, 0x81, 0x5a, 0xa4, 0x5c, 0x9b, 0xa6, 0xa9, 0xab, 0x13,
+	0x53, 0x24, 0x76, 0x13, 0xfb, 0xc4, 0x19, 0xc5, 0x9e, 0x63, 0x66, 0xc6, 0x89, 0xf3, 0x06, 0x3c,
+	0x01, 0x0b, 0x16, 0xac, 0x79, 0x0d, 0x76, 0x5d, 0x56, 0x42, 0x6c, 0x11, 0x64, 0xc3, 0x6b, 0xa0,
+	0x73, 0x99, 0x9b, 0x2f, 0x49, 0xe9, 0xee, 0xfc, 0xf7, 0xcb, 0x7c, 0xf3, 0xff, 0xe7, 0x00, 0xf4,
+	0x68, 0xfb, 0x62, 0x6d, 0xe0, 0xd1, 0x80, 0x22, 0x95, 0x9d, 0xef, 0x7f, 0xde, 0x75, 0x82, 0xf3,
+	0xe1, 0xe9, 0x5a, 0x9b, 0xf6, 0xd7, 0xbb, 0xb4, 0x4b, 0xd7, 0xb9, 0xf0, 0x74, 0x78, 0xc6, 0x29,
+	0x4e, 0xf0, 0x93, 0x30, 0xba, 0x3f, 0x1f, 0x38, 0x7d, 0xe2, 0x07, 0x76, 0x7f, 0x20, 0x18, 0xd6,
+	0x2f, 0x19, 0xd0, 0x5f, 0xd0, 0xf6, 0x45, 0x73, 0x10, 0x38, 0xd4, 0xf5, 0xd1, 0x26, 0xe8, 0x07,
+	0x9e, 0xed, 0x0e, 0x7b, 0xb6, 0xe7, 0x04, 0xd7, 0xa6, 0x52, 0x55, 0x6a, 0x95, 0xc6, 0xc2, 0x1a,
+	0x8f, 0x9b, 0x10, 0xe0, 0xa4, 0x16, 0xb2, 0x40, 0x3d, 0xa6, 0x1d, 0x62, 0x66, 0xb8, 0x76, 0x45,
+	0x68, 0x33, 0xaf, 0x8c, 0x8b, 0xb9, 0x0c, 0xd5, 0x20, 0xff, 0x8a, 0xf6, 0x9c, 0xf6, 0xb5, 0x99,
+	0xe5, 0x5a, 0x86, 0xd0, 0xfa, 0xde, 0x76, 0x02, 0xc1, 0xc7, 0x52, 0x8e, 0x1e, 0x40, 0x71, 0x9f,
+	0x7a, 0x57, 0xb6, 0xd7, 0x69, 0x51, 0x53, 0xad, 0x2a, 0xb5, 0x22, 0x8e, 0x19, 0xa8, 0x06, 0xf3,
+	0x2d, 0xfb, 0xb4, 0x47, 0x76, 0xc9, 0xd9, 0xce, 0xb9, 0xed, 0x76, 0x49, 0xc7, 0xcc, 0x55, 0x95,
+	0x9a, 0x86, 0xc7, 0xd9, 0x68, 0x11, 0x72, 0x07, 0x1e, 0x1d, 0x0e, 0xcc, 0x7c, 0x55, 0xa9, 0x95,
+	0xb1, 0x20, 0x50, 0x1d, 0xb4, 0x93, 0x73, 0xdb, 0xeb, 0x38, 0x6e, 0xd7, 0x2c, 0x24, 0xf3, 0x0d,
+	0xb9, 0x38, 0x92, 0x5b, 0x7f, 0x28, 0x50, 0x64, 0x65, 0x70, 0xcf, 0xcc, 0x1f, 0x3f, 0xf0, 0xa6,
+	0xa8, 0x58, 0x10, 0x2c, 0xdb, 0x13, 0xe2, 0x5d, 0x3a, 0x6d, 0x72, 0xb8, 0xcb, 0x1b, 0x50, 0xc4,
+	0x31, 0x03, 0x99, 0x50, 0x78, 0x4d, 0x3c, 0xdf, 0xa1, 0x2e, 0x2f, 0x5b, 0xc5, 0x21, 0xc9, 0xbc,
+	0xbd, 0xb6, 0x7b, 0x4e, 0x87, 0x57, 0xa8, 0x61, 0x41, 0xc4, 0x39, 0xe7, 0x66, 0xe5, 0x9c, 0xbf,
+	0x3d, 0x67, 0x54, 0x05, 0xbd, 0xe9, 0x39, 0x5d, 0xc7, 0x15, 0xb9, 0x16, 0x78, 0xd4, 0x24, 0xcb,
+	0xfa, 0x5d, 0x85, 0x02, 0x26, 0x3f, 0x0e, 0x89, 0x1f, 0xb0, 0xec, 0xe5, 0xf1, 0x70, 0x57, 0xd6,
+	0x15, 0x33, 0xd0, 0x66, 0xa2, 0x7c, 0x5e, 0x9b, 0xde, 0x98, 0x8f, 0x3f, 0x2e, 0x67, 0x6f, 0xab,
+	0x6f, 0xfe, 0x5a, 0x99, 0xc3, 0x89, 0x36, 0xad, 0x42, 0xfe, 0x98, 0x04, 0xe7, 0xb4, 0x23, 0x3f,
+	0x74, 0x49, 0x58, 0x08, 0x1e, 0x96, 0x32, 0xf4, 0x08, 0x54, 0x66, 0xc2, 0xab, 0xd7, 0x43, 0x80,
+	0x31, 0x8e, 0x8c, 0x2e, 0xfd, 0x72, 0x25, 0xb4, 0x01, 0xf9, 0xef, 0x5c, 0xa6, 0xc1, 0xdb, 0xa2,
+	0x37, 0x3e, 0x10, 0xea, 0x82, 0x97, 0x36, 0x90, 0x8a, 0xe8, 0x09, 0xc0, 0x01, 0x09, 0x5a, 0x23,
+	0x97, 0x47, 0xc9, 0x73, 0xb3, 0x0f, 0x25, 0x8c, 0x23, 0x7e, 0xda, 0x34, 0x61, 0x80, 0x0e, 0xa1,
+	0x72, 0x40, 0x02, 0x06, 0x4e, 0xc7, 0xed, 0xbe, 0x70, 0xfc, 0x80, 0x37, 0x52, 0x6f, 0x7c, 0x1c,
+	0xb9, 0x48, 0xc8, 0xd2, 0x6e, 0xc6, 0x0c, 0xd1, 0x17, 0x50, 0x38, 0x20, 0xc1, 0xb6, 0xe3, 0x76,
+	0x4c, 0x8d, 0xfb, 0x58, 0x8c, 0x7c, 0x30, 0x66, 0xda, 0x38, 0x54, 0x45, 0x18, 0x16, 0x8e, 0x08,
+	0x19, 0xc4, 0x7d, 0x66, 0xf6, 0x45, 0x6e, 0xbf, 0x2c, 0xec, 0x27, 0xc4, 0x69, 0x4f, 0x93, 0xe6,
+	0xac, 0x28, 0xc6, 0xc4, 0xa4, 0x4f, 0x03, 0xc2, 0xfb, 0x02, 0xc9, 0xa2, 0xd2, 0xb2, 0xb1, 0xa2,
+	0xd2, 0x42, 0xeb, 0x4f, 0x15, 0x34, 0x4c, 0xfc, 0x01, 0x75, 0x7d, 0x72, 0x07, 0x88, 0x62, 0x3c,
+	0x64, 0x6e, 0xc1, 0xc3, 0x22, 0xe4, 0xf6, 0x3c, 0x8f, 0x7a, 0x1c, 0x34, 0x25, 0x2c, 0x08, 0xf4,
+	0x29, 0x14, 0x5e, 0x92, 0x2b, 0x5e, 0xbb, 0x3a, 0x15, 0x7e, 0x38, 0x94, 0xa3, 0xcf, 0x24, 0xa0,
+	0x04, 0x42, 0x50, 0x12, 0x50, 0x22, 0xcd, 0x14, 0xa2, 0x1a, 0x11, 0xa2, 0xf2, 0xc9, 0x6f, 0x12,
+	0x22, 0x2a, 0x65, 0x11, 0x42, 0xea, 0x69, 0x0a, 0x52, 0x02, 0x0f, 0xe6, 0x24, 0xa4, 0x52, 0xb6,
+	0x49, 0x4c, 0x3d, 0x9f, 0xc0, 0x94, 0xc0, 0xc3, 0x83, 0xe9, 0x98, 0x4a, 0xf9, 0x19, 0x07, 0xd5,
+	0x56, 0x0c, 0x2a, 0x01, 0x8a, 0xa5, 0x31, 0x50, 0xa5, 0xac, 0x23, 0x54, 0x9d, 0x4c, 0x43, 0x95,
+	0x00, 0xc1, 0xca, 0x4c, 0x54, 0xa5, 0x5c, 0x4d, 0x81, 0xd5, 0xf3, 0x09, 0x58, 0xe9, 0xc9, 0xba,
+	0xc6, 0x61, 0x95, 0xae, 0x6b, 0x0c, 0x57, 0x3f, 0x29, 0x62, 0x1d, 0x85, 0xf3, 0x89, 0xcd, 0xdc,
+	0x91, 0x2b, 0x61, 0x55, 0xc2, 0x82, 0xb8, 0x63, 0xe6, 0x22, 0x50, 0x31, 0xbd, 0xf2, 0xcd, 0x6c,
+	0x35, 0x5b, 0x2b, 0x61, 0x7e, 0x46, 0x1b, 0x50, 0x90, 0x1b, 0x6e, 0x72, 0xe2, 0x48, 0x41, 0xd8,
+	0x2b, 0x49, 0x5a, 0x5f, 0x43, 0x29, 0x99, 0x30, 0xaa, 0x43, 0x1e, 0x13, 0x7f, 0xd8, 0x0b, 0x78,
+	0x2e, 0x7a, 0x88, 0x63, 0xc1, 0x0b, 0xa1, 0x22, 0x28, 0xeb, 0x1b, 0x58, 0x98, 0x98, 0x32, 0x33,
+	0x6a, 0x31, 0x20, 0x8b, 0xe9, 0x15, 0xaf, 0xa2, 0x84, 0xd9, 0xd1, 0xb2, 0x01, 0x4d, 0xe2, 0x49,
+	0xee, 0x8b, 0xa1, 0xd8, 0x3e, 0x39, 0x2c, 0x08, 0xb4, 0x05, 0x7a, 0x12, 0x50, 0x99, 0x6a, 0xb6,
+	0xa6, 0x37, 0xca, 0xf1, 0x6a, 0x6d, 0x8d, 0x5c, 0x99, 0x5a, 0x52, 0xcf, 0x7a, 0x0a, 0x4b, 0x53,
+	0x47, 0x18, 0x7a, 0x08, 0xd9, 0xd6, 0xc8, 0x95, 0x15, 0x4e, 0xf5, 0xc3, 0xe4, 0x56, 0x13, 0xee,
+	0x4d, 0x87, 0xeb, 0x78, 0x42, 0xca, 0x3b, 0x26, 0xf4, 0x04, 0x0a, 0x52, 0x3a, 0xfb, 0x93, 0xef,
+	0x78, 0xc4, 0x0e, 0x48, 0xa7, 0xe9, 0x86, 0x9f, 0x3c, 0x62, 0x58, 0x3f, 0x2b, 0x50, 0x4e, 0x6d,
+	0x83, 0x19, 0x5e, 0xbe, 0x04, 0x6d, 0x87, 0xf6, 0xfb, 0x4e, 0xd0, 0x3a, 0x91, 0xfb, 0x6c, 0x71,
+	0x2d, 0xbe, 0x11, 0xb5, 0xc2, 0x93, 0xcc, 0x30, 0xd2, 0x45, 0x5f, 0x41, 0xf1, 0x78, 0x18, 0xd8,
+	0x02, 0x40, 0x59, 0x5e, 0x93, 0xdc, 0x41, 0x7b, 0xa3, 0xc0, 0xb3, 0x43, 0x59, 0xb8, 0x0c, 0x23,
+	0x5d, 0xcb, 0x80, 0x4a, 0x7a, 0xa6, 0x58, 0xbf, 0x29, 0x7c, 0x0c, 0x24, 0x06, 0x76, 0x1a, 0xce,
+	0xca, 0x38, 0x9c, 0xa3, 0x6b, 0x47, 0x26, 0x79, 0xed, 0x88, 0x2e, 0x0a, 0xd9, 0x59, 0x17, 0x05,
+	0xf5, 0xff, 0x5d, 0x14, 0x72, 0x93, 0x17, 0x85, 0x7d, 0x98, 0x1f, 0x9b, 0x27, 0xef, 0x75, 0x23,
+	0xb0, 0x1e, 0x83, 0x39, 0x6b, 0x59, 0xdd, 0x5e, 0xbb, 0xf5, 0x08, 0x3e, 0x9a, 0x39, 0x90, 0x50,
+	0x05, 0x32, 0xcd, 0x23, 0x6e, 0xa3, 0xe1, 0x4c, 0xf3, 0xc8, 0xda, 0x82, 0xa5, 0xa9, 0x2b, 0xec,
+	0x8e, 0x18, 0x35, 0xb8, 0x37, 0x7d, 0x44, 0x4d, 0x04, 0xf8, 0x57, 0x09, 0x47, 0x00, 0xda, 0x00,
+	0x8d, 0xa9, 0x72, 0x34, 0x2a, 0xb7, 0xb5, 0x21, 0x52, 0x63, 0xfd, 0x7e, 0x66, 0xfb, 0x3b, 0xd4,
+	0x3d, 0xeb, 0x39, 0xed, 0x80, 0x37, 0x4f, 0xc3, 0x49, 0x16, 0x5a, 0x85, 0xf2, 0x33, 0xdb, 0x7f,
+	0xe5, 0x91, 0x4b, 0x01, 0x3c, 0xfe, 0x6d, 0x35, 0x9c, 0x66, 0xa2, 0xc7, 0x50, 0x8c, 0x80, 0x2a,
+	0x87, 0xd9, 0x6d, 0x20, 0x8e, 0x95, 0xdf, 0xfd, 0xea, 0x6c, 0x9d, 0x40, 0x39, 0x05, 0x6c, 0x36,
+	0xa5, 0x2e, 0xc8, 0xb5, 0xfc, 0x99, 0xd8, 0x91, 0x4d, 0x59, 0xff, 0xc2, 0x19, 0xc8, 0x3a, 0xf8,
+	0x99, 0x35, 0xda, 0x23, 0x83, 0x9e, 0xdd, 0x26, 0x2d, 0x2a, 0x17, 0x79, 0xcc, 0xa8, 0x7f, 0x92,
+	0x7a, 0x5a, 0xa0, 0x02, 0x1f, 0x7c, 0xc6, 0x1c, 0x2a, 0x42, 0x0e, 0xb3, 0xb0, 0x86, 0x52, 0x7f,
+	0x28, 0xda, 0xca, 0x1f, 0x0c, 0x65, 0x28, 0xee, 0x8d, 0xda, 0xbd, 0xa1, 0xef, 0x5c, 0x12, 0x63,
+	0x0e, 0x01, 0xe4, 0x19, 0x74, 0x49, 0xc7, 0x50, 0xea, 0xab, 0x00, 0xf1, 0xbb, 0x01, 0x69, 0xa0,
+	0x32, 0xca, 0x98, 0x43, 0x25, 0xd0, 0xf6, 0x6d, 0x3f, 0xd8, 0xb7, 0x9d, 0x9e, 0xa1, 0xd4, 0x57,
+	0xe2, 0x9f, 0x81, 0xe9, 0xbc, 0xa4, 0x2e, 0x11, 0xd1, 0xb6, 0xaf, 0x59, 0x60, 0xa5, 0xfe, 0xab,
+	0x12, 0x5e, 0x4d, 0x98, 0x9c, 0x05, 0x16, 0x71, 0xc4, 0x1f, 0x6b, 0x28, 0xa8, 0x92, 0xdc, 0xf8,
+	0x46, 0x06, 0xa1, 0xf1, 0x0d, 0x6e, 0x64, 0x19, 0x2f, 0x0d, 0x1f, 0x43, 0x45, 0x7a, 0xb4, 0x9d,
+	0x8d, 0x1c, 0x5a, 0x9a, 0xb2, 0x73, 0x8d, 0x3c, 0x9a, 0x07, 0x5d, 0x3e, 0x6a, 0xb8, 0x51, 0x01,
+	0x2d, 0x40, 0x59, 0x32, 0x64, 0x7c, 0x6d, 0xfb, 0xdb, 0xb7, 0xff, 0x2c, 0x2b, 0x6f, 0x6e, 0x96,
+	0x95, 0xb7, 0x37, 0xcb, 0xca, 0xdf, 0x37, 0xcb, 0xca, 0x0f, 0xc9, 0xe7, 0x5e, 0xdf, 0x0e, 0x3c,
+	0x67, 0x44, 0xf9, 0x0f, 0x1b, 0x12, 0x2e, 0x59, 0x1f, 0x5c, 0x74, 0xd7, 0x07, 0xa7, 0xeb, 0xcc,
+	0xc5, 0x69, 0x9e, 0x3f, 0xf2, 0x36, 0xff, 0x0b, 0x00, 0x00, 0xff, 0xff, 0xdf, 0x70, 0x86, 0x47,
+	0x38, 0x0e, 0x00, 0x00,
 }
 
 func (m *LockOptions) Marshal() (dAtA []byte, err error) {
@@ -1567,6 +1738,16 @@ func (m *LockOptions) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.Sharding != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.Sharding))
+		i--
+		dAtA[i] = 0x38
+	}
+	if m.Group != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.Group))
+		i--
+		dAtA[i] = 0x30
+	}
 	if m.TableDefChanged {
 		i--
 		if m.TableDefChanged {
@@ -1626,6 +1807,21 @@ func (m *LockTable) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.OriginTable != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.OriginTable))
+		i--
+		dAtA[i] = 0x38
+	}
+	if m.Sharding != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.Sharding))
+		i--
+		dAtA[i] = 0x30
+	}
+	if m.Group != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.Group))
+		i--
+		dAtA[i] = 0x28
+	}
 	if m.Valid {
 		i--
 		if m.Valid {
@@ -2226,6 +2422,20 @@ func (m *UnlockRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if len(m.Mutations) > 0 {
+		for iNdEx := len(m.Mutations) - 1; iNdEx >= 0; iNdEx-- {
+			{
+				size, err := m.Mutations[iNdEx].MarshalToSizedBuffer(dAtA[:i])
+				if err != nil {
+					return 0, err
+				}
+				i -= size
+				i = encodeVarintLock(dAtA, i, uint64(size))
+			}
+			i--
+			dAtA[i] = 0x1a
+		}
+	}
 	{
 		size, err := m.CommitTS.MarshalToSizedBuffer(dAtA[:i])
 		if err != nil {
@@ -2297,6 +2507,21 @@ func (m *GetBindRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.OriginTable != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.OriginTable))
+		i--
+		dAtA[i] = 0x28
+	}
+	if m.Sharding != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.Sharding))
+		i--
+		dAtA[i] = 0x20
+	}
+	if m.Group != 0 {
+		i = encodeVarintLock(dAtA, i, uint64(m.Group))
+		i--
+		dAtA[i] = 0x18
+	}
 	if m.Table != 0 {
 		i = encodeVarintLock(dAtA, i, uint64(m.Table))
 		i--
@@ -2568,6 +2793,57 @@ func (m *Result) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *ExtraMutation) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *ExtraMutation) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *ExtraMutation) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.ReplaceTo) > 0 {
+		i -= len(m.ReplaceTo)
+		copy(dAtA[i:], m.ReplaceTo)
+		i = encodeVarintLock(dAtA, i, uint64(len(m.ReplaceTo)))
+		i--
+		dAtA[i] = 0x1a
+	}
+	if m.Skip {
+		i--
+		if m.Skip {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x10
+	}
+	if len(m.Key) > 0 {
+		i -= len(m.Key)
+		copy(dAtA[i:], m.Key)
+		i = encodeVarintLock(dAtA, i, uint64(len(m.Key)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
 func encodeVarintLock(dAtA []byte, offset int, v uint64) int {
 	offset -= sovLock(v)
 	base := offset
@@ -2601,6 +2877,12 @@ func (m *LockOptions) Size() (n int) {
 	if m.TableDefChanged {
 		n += 2
 	}
+	if m.Group != 0 {
+		n += 1 + sovLock(uint64(m.Group))
+	}
+	if m.Sharding != 0 {
+		n += 1 + sovLock(uint64(m.Sharding))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -2626,6 +2908,15 @@ func (m *LockTable) Size() (n int) {
 	if m.Valid {
 		n += 2
 	}
+	if m.Group != 0 {
+		n += 1 + sovLock(uint64(m.Group))
+	}
+	if m.Sharding != 0 {
+		n += 1 + sovLock(uint64(m.Sharding))
+	}
+	if m.OriginTable != 0 {
+		n += 1 + sovLock(uint64(m.OriginTable))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -2853,6 +3144,12 @@ func (m *UnlockRequest) Size() (n int) {
 	}
 	l = m.CommitTS.Size()
 	n += 1 + l + sovLock(uint64(l))
+	if len(m.Mutations) > 0 {
+		for _, e := range m.Mutations {
+			l = e.Size()
+			n += 1 + l + sovLock(uint64(l))
+		}
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -2884,6 +3181,15 @@ func (m *GetBindRequest) Size() (n int) {
 	if m.Table != 0 {
 		n += 1 + sovLock(uint64(m.Table))
 	}
+	if m.Group != 0 {
+		n += 1 + sovLock(uint64(m.Group))
+	}
+	if m.Sharding != 0 {
+		n += 1 + sovLock(uint64(m.Sharding))
+	}
+	if m.OriginTable != 0 {
+		n += 1 + sovLock(uint64(m.OriginTable))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -2991,6 +3297,29 @@ func (m *Result) Size() (n int) {
 	return n
 }
 
+func (m *ExtraMutation) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.Key)
+	if l > 0 {
+		n += 1 + l + sovLock(uint64(l))
+	}
+	if m.Skip {
+		n += 2
+	}
+	l = len(m.ReplaceTo)
+	if l > 0 {
+		n += 1 + l + sovLock(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func sovLock(x uint64) (n int) {
 	return (math_bits.Len64(x|1) + 6) / 7
 }
@@ -3135,6 +3464,44 @@ func (m *LockOptions) Unmarshal(dAtA []byte) error {
 				}
 			}
 			m.TableDefChanged = bool(v != 0)
+		case 6:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Group", wireType)
+			}
+			m.Group = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Group |= uint32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 7:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Sharding", wireType)
+			}
+			m.Sharding = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Sharding |= Sharding(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipLock(dAtA[iNdEx:])
@@ -3276,6 +3643,63 @@ func (m *LockTable) Unmarshal(dAtA []byte) error {
 				}
 			}
 			m.Valid = bool(v != 0)
+		case 5:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Group", wireType)
+			}
+			m.Group = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Group |= uint32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 6:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Sharding", wireType)
+			}
+			m.Sharding = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Sharding |= Sharding(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 7:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field OriginTable", wireType)
+			}
+			m.OriginTable = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.OriginTable |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipLock(dAtA[iNdEx:])
@@ -4912,6 +5336,40 @@ func (m *UnlockRequest) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Mutations", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthLock
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthLock
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Mutations = append(m.Mutations, ExtraMutation{})
+			if err := m.Mutations[len(m.Mutations)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipLock(dAtA[iNdEx:])
@@ -5065,6 +5523,63 @@ func (m *GetBindRequest) Unmarshal(dAtA []byte) error {
 					break
 				}
 			}
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Group", wireType)
+			}
+			m.Group = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Group |= uint32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 4:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Sharding", wireType)
+			}
+			m.Sharding = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Sharding |= Sharding(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 5:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field OriginTable", wireType)
+			}
+			m.OriginTable = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.OriginTable |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipLock(dAtA[iNdEx:])
@@ -5656,6 +6171,145 @@ func (m *Result) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *ExtraMutation) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowLock
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ExtraMutation: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ExtraMutation: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Key", wireType)
+			}
+			var byteLen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				byteLen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if byteLen < 0 {
+				return ErrInvalidLengthLock
+			}
+			postIndex := iNdEx + byteLen
+			if postIndex < 0 {
+				return ErrInvalidLengthLock
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Key = append(m.Key[:0], dAtA[iNdEx:postIndex]...)
+			if m.Key == nil {
+				m.Key = []byte{}
+			}
+			iNdEx = postIndex
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Skip", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.Skip = bool(v != 0)
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ReplaceTo", wireType)
+			}
+			var byteLen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowLock
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				byteLen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if byteLen < 0 {
+				return ErrInvalidLengthLock
+			}
+			postIndex := iNdEx + byteLen
+			if postIndex < 0 {
+				return ErrInvalidLengthLock
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.ReplaceTo = append(m.ReplaceTo[:0], dAtA[iNdEx:postIndex]...)
+			if m.ReplaceTo == nil {
+				m.ReplaceTo = []byte{}
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipLock(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthLock
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func skipLock(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/pkg/sql/colexec/lockop/lock_op.go b/pkg/sql/colexec/lockop/lock_op.go
index 485b149f2..2988c5e8a 100644
--- a/pkg/sql/colexec/lockop/lock_op.go
+++ b/pkg/sql/colexec/lockop/lock_op.go
@@ -326,6 +326,8 @@ func LockRows(
 	vec *vector.Vector,
 	pkType types.Type,
 	lockMode lock.LockMode,
+	sharding lock.Sharding,
+	group uint32,
 ) error {
 	if !proc.TxnOperator.Txn().IsPessimistic() {
 		return nil
@@ -336,7 +338,9 @@ func LockRows(
 
 	opts := DefaultLockOptions(parker).
 		WithLockTable(false, false).
+		WithLockSharding(sharding).
 		WithLockMode(lockMode).
+		WithLockGroup(group).
 		WithFetchLockRowsFunc(GetFetchRowsFunc(pkType))
 	_, defChanged, refreshTS, err := doLock(
 		proc.Ctx,
@@ -419,6 +423,8 @@ func doLock(
 		Policy:          lock.WaitPolicy_Wait,
 		Mode:            opts.mode,
 		TableDefChanged: opts.changeDef,
+		Sharding:        opts.sharding,
+		Group:           opts.group,
 	}
 	if txn.Mirror {
 		options.ForwardTo = txn.LockService
@@ -527,6 +533,18 @@ func DefaultLockOptions(parker *types.Packer) LockOptions {
 	}
 }
 
+// WithLockSharding set lock sharding
+func (opts LockOptions) WithLockSharding(sharding lock.Sharding) LockOptions {
+	opts.sharding = sharding
+	return opts
+}
+
+// WithLockGroup set lock group
+func (opts LockOptions) WithLockGroup(group uint32) LockOptions {
+	opts.group = group
+	return opts
+}
+
 // WithLockMode set lock mode, Exclusive or Shared
 func (opts LockOptions) WithLockMode(mode lock.LockMode) LockOptions {
 	opts.mode = mode
diff --git a/pkg/sql/colexec/lockop/types.go b/pkg/sql/colexec/lockop/types.go
index e02639dc4..3ec456c82 100644
--- a/pkg/sql/colexec/lockop/types.go
+++ b/pkg/sql/colexec/lockop/types.go
@@ -49,6 +49,8 @@ type FetchLockRowsFunc func(
 type LockOptions struct {
 	maxCountPerLock          int
 	mode                     lock.LockMode
+	sharding                 lock.Sharding
+	group                    uint32
 	lockTable                bool
 	changeDef                bool
 	parker                   *types.Packer
diff --git a/pkg/sql/compile/compile.go b/pkg/sql/compile/compile.go
index 0b3cf67be..3e9067701 100644
--- a/pkg/sql/compile/compile.go
+++ b/pkg/sql/compile/compile.go
@@ -91,8 +91,7 @@ const (
 )
 
 var (
-	ncpu = runtime.NumCPU()
-
+	ncpu           = runtime.NumCPU()
 	ctxCancelError = context.Canceled.Error()
 )
 
diff --git a/pkg/sql/compile/ddl.go b/pkg/sql/compile/ddl.go
index 5b586783f..7750a005c 100644
--- a/pkg/sql/compile/ddl.go
+++ b/pkg/sql/compile/ddl.go
@@ -2604,8 +2604,9 @@ func lockRows(
 	proc *process.Process,
 	rel engine.Relation,
 	vec *vector.Vector,
-	lockMode lock.LockMode) error {
-
+	lockMode lock.LockMode,
+	sharding lock.Sharding,
+	group uint32) error {
 	if vec == nil || vec.Length() == 0 {
 		panic("lock rows is empty")
 	}
@@ -2619,7 +2620,9 @@ func lockRows(
 		id,
 		vec,
 		*vec.GetType(),
-		lockMode)
+		lockMode,
+		sharding,
+		group)
 	return err
 }
 
@@ -2708,13 +2711,17 @@ func lockMoDatabase(c *Compile, dbName string) error {
 	if err != nil {
 		return err
 	}
-	if err := lockRows(c.e, c.proc, dbRel, vec, lock.LockMode_Exclusive); err != nil {
+	if err := lockRows(c.e, c.proc, dbRel, vec, lock.LockMode_Exclusive, lock.Sharding_ByRow, c.proc.SessionInfo.AccountId); err != nil {
 		return err
 	}
 	return nil
 }
 
-func lockMoTable(c *Compile, dbName string, tblName string, lockMode lock.LockMode) error {
+func lockMoTable(
+	c *Compile,
+	dbName string,
+	tblName string,
+	lockMode lock.LockMode) error {
 	dbRel, err := getRelFromMoCatalog(c, catalog.MO_TABLES)
 	if err != nil {
 		return err
@@ -2724,7 +2731,8 @@ func lockMoTable(c *Compile, dbName string, tblName string, lockMode lock.LockMo
 		return err
 	}
 	defer vec.Free(c.proc.Mp())
-	if err := lockRows(c.e, c.proc, dbRel, vec, lockMode); err != nil {
+
+	if err := lockRows(c.e, c.proc, dbRel, vec, lockMode, lock.Sharding_ByRow, c.proc.SessionInfo.AccountId); err != nil {
 		return err
 	}
 	return nil
diff --git a/pkg/txn/client/operator.go b/pkg/txn/client/operator.go
index eabdc2c97..b3b09d49c 100644
--- a/pkg/txn/client/operator.go
+++ b/pkg/txn/client/operator.go
@@ -584,7 +584,8 @@ func (tc *txnOperator) IsOpenLog() bool {
 
 func (tc *txnOperator) doAddLockTableLocked(value lock.LockTable) error {
 	for _, l := range tc.mu.lockTables {
-		if l.Table == value.Table {
+		if l.Group == value.Group &&
+			l.Table == value.Table {
 			if l.Changed(value) {
 				return moerr.NewDeadLockDetectedNoCtx()
 			}
diff --git a/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go b/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go
index a68b957b4..43f10aae9 100644
--- a/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go
+++ b/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go
@@ -427,6 +427,7 @@ func (c *DashboardCreator) initTxnLockDurationRow() dashboard.Option {
 				c.getMetricWithFilter(`mo_txn_unlock_duration_seconds_bucket`, `type="total"`),
 				c.getMetricWithFilter(`mo_txn_unlock_duration_seconds_bucket`, `type="btree-get-lock"`),
 				c.getMetricWithFilter(`mo_txn_unlock_duration_seconds_bucket`, `type="btree-total"`),
+				c.getMetricWithFilter(`mo_txn_unlock_duration_seconds_bucket`, `type="worker-handle"`),
 			},
 			[]string{
 				"lock-total",
@@ -434,6 +435,7 @@ func (c *DashboardCreator) initTxnLockDurationRow() dashboard.Option {
 				"unlock-total",
 				"unlock-btree-get-lock",
 				"unlock-btree-total",
+				"worker-handle",
 			},
 			[]float64{0.50, 0.8, 0.90, 0.99},
 			[]float32{3, 3, 3, 3},
diff --git a/pkg/util/metric/v2/txn.go b/pkg/util/metric/v2/txn.go
index 36b4a713f..314f8c7c9 100644
--- a/pkg/util/metric/v2/txn.go
+++ b/pkg/util/metric/v2/txn.go
@@ -215,6 +215,7 @@ var (
 	TxnUnlockDurationHistogram             = txnUnlockDurationHistogram.WithLabelValues("total")
 	TxnUnlockBtreeGetLockDurationHistogram = txnUnlockDurationHistogram.WithLabelValues("btree-get-lock")
 	TxnUnlockBtreeTotalDurationHistogram   = txnUnlockDurationHistogram.WithLabelValues("btree-total")
+	TxnLockWorkerHandleDurationHistogram   = txnUnlockDurationHistogram.WithLabelValues("worker-handle")
 
 	TxnLockWaitersTotalHistogram = prometheus.NewHistogram(
 		prometheus.HistogramOpts{
diff --git a/proto/lock.proto b/proto/lock.proto
index 4c5a3bee7..df729596a 100644
--- a/proto/lock.proto
+++ b/proto/lock.proto
@@ -41,25 +41,35 @@ enum WaitPolicy {
   FastFail = 1;
 }
 
+enum Sharding {
+  None   = 0;
+  ByRow  = 1;
+}
+
 // LockOptions lock options
 message LockOptions {
-  Granularity Granularity     = 1;
-  LockMode    Mode            = 2;
-  WaitPolicy  Policy          = 3;
-  string      ForwardTo       = 4;
-  bool        TableDefChanged = 5;
+  Granularity Granularity       = 1;
+  LockMode    Mode              = 2;
+  WaitPolicy  Policy            = 3;
+  string      ForwardTo         = 4;
+  bool        TableDefChanged   = 5;
+  uint32      Group             = 6;
+  Sharding    Sharding          = 7;
 }
 
 // LockTable describes which CN manages a Table's Locks.
 message LockTable {
   // Table table id
-  uint64 Table     = 1;
+  uint64   Table       = 1;
   // ServiceID lock service id, a cn node will only run one instance of LockService
-  string ServiceID = 2;
+  string   ServiceID   = 2;
   // Version will incr if CN-Table bind changed.
-  uint64 Version   = 3;
+  uint64   Version     = 3;
   // Valid false if the service is disabled, and no new service bind this table
-  bool   Valid     = 4;
+  bool     Valid       = 4;
+  uint32   Group       = 5;
+  Sharding Sharding    = 6;
+  uint64   OriginTable = 7;
 }
 
 // Method lock table operations
@@ -164,10 +174,11 @@ message WaitTxn {
 
 // UnlockRequest unlock lock on remote lock service request. CN -> CN
 message UnlockRequest {
-  bytes               TxnID     = 1;
+  bytes                  TxnID     = 1;
   // CommitTS is the commit timestamp of the current txn. Empty if txn is
   // roll backed
-  timestamp.Timestamp CommitTS  = 2 [(gogoproto.nullable) = false];
+  timestamp.Timestamp    CommitTS  = 2 [(gogoproto.nullable) = false];
+  repeated ExtraMutation Mutations = 3 [(gogoproto.nullable) = false];
 }
 
 // UnlockResponse unlock lock on remote lock service response. CN -> CN
@@ -177,8 +188,11 @@ message UnlockResponse {
 
 // GetBindRequest get bind request from allocator request. CN -> TN
 message GetBindRequest {
-  string ServiceID = 1;
-  uint64 Table     = 2;
+  string   ServiceID   = 1;
+  uint64   Table       = 2;
+  uint32   Group       = 3;
+  Sharding Sharding    = 4;
+  uint64   OriginTable = 5;
 }
 
 // GetBindResponse get bind request from allocator response. TN -> CN
@@ -221,4 +235,10 @@ message Result {
   timestamp.Timestamp Timestamp       = 4 [(gogoproto.nullable) = false];
   // TableDefChanged conflict with ddl lock, need rebuild plan to get new table def
   bool                TableDefChanged = 5;
+}
+
+message ExtraMutation {
+  bytes key       = 1;
+  bool  skip      = 2;
+  bytes replaceTo = 3;
 }
\ No newline at end of file


Test patch:
diff --git a/pkg/lockservice/lock_table_allocator_test.go b/pkg/lockservice/lock_table_allocator_test.go
index 641ca01bc..05b6cbc93 100644
--- a/pkg/lockservice/lock_table_allocator_test.go
+++ b/pkg/lockservice/lock_table_allocator_test.go
@@ -42,8 +42,8 @@ func TestGetWithNoBind(t *testing.T) {
 		time.Hour,
 		func(a *lockTableAllocator) {
 			assert.Equal(t,
-				pb.LockTable{Valid: true, ServiceID: "s1", Table: 1, Version: 1},
-				a.Get("s1", 1))
+				pb.LockTable{Valid: true, ServiceID: "s1", Table: 1, OriginTable: 1, Version: 1},
+				a.Get("s1", 0, 1, 0, pb.Sharding_None))
 		})
 }
 
@@ -53,10 +53,10 @@ func TestGetWithAlreadyBind(t *testing.T) {
 		time.Hour,
 		func(a *lockTableAllocator) {
 			// register s1 first
-			a.Get("s1", 1)
+			a.Get("s1", 0, 1, 0, pb.Sharding_None)
 			assert.Equal(t,
-				pb.LockTable{Valid: true, ServiceID: "s1", Table: 1, Version: 1},
-				a.Get("s2", 1))
+				pb.LockTable{Valid: true, ServiceID: "s1", Table: 1, OriginTable: 1, Version: 1},
+				a.Get("s2", 0, 1, 0, pb.Sharding_None))
 		})
 }
 
@@ -66,11 +66,11 @@ func TestGetWithBindInvalid(t *testing.T) {
 		time.Hour,
 		func(a *lockTableAllocator) {
 			// register s1 first
-			a.Get("s1", 1)
+			a.Get("s1", 0, 1, 0, pb.Sharding_None)
 			a.disableTableBinds(a.getServiceBinds("s1"))
 			assert.Equal(t,
-				pb.LockTable{Valid: true, ServiceID: "s2", Table: 1, Version: 2},
-				a.Get("s2", 1))
+				pb.LockTable{Valid: true, ServiceID: "s2", Table: 1, OriginTable: 1, Version: 2},
+				a.Get("s2", 0, 1, 0, pb.Sharding_None))
 		})
 }
 
@@ -80,16 +80,16 @@ func TestGetWithBindAndServiceBothInvalid(t *testing.T) {
 		time.Hour,
 		func(a *lockTableAllocator) {
 			// invalid table 1 bind
-			a.Get("s1", 1)
+			a.Get("s1", 0, 1, 0, pb.Sharding_None)
 			a.disableTableBinds(a.getServiceBinds("s1"))
 
 			// invalid s2
-			a.Get("s2", 2)
+			a.Get("s2", 0, 2, 0, pb.Sharding_None)
 			a.getServiceBinds("s2").disable()
 
 			assert.Equal(t,
-				pb.LockTable{Valid: false, ServiceID: "s1", Table: 1, Version: 1},
-				a.Get("s2", 1))
+				pb.LockTable{Valid: false, ServiceID: "s1", Table: 1, OriginTable: 1, Version: 1},
+				a.Get("s2", 0, 1, 0, pb.Sharding_None))
 		})
 }
 
@@ -99,7 +99,7 @@ func TestCheckTimeoutServiceTask(t *testing.T) {
 		time.Millisecond,
 		func(a *lockTableAllocator) {
 			// create s1 bind
-			a.Get("s1", 1)
+			a.Get("s1", 0, 1, 0, pb.Sharding_None)
 
 			// wait bind timeout
 			for {
@@ -109,10 +109,10 @@ func TestCheckTimeoutServiceTask(t *testing.T) {
 					continue
 				}
 				a.mu.Lock()
-				if len(a.mu.lockTables) > 0 {
+				if len(a.getLockTablesLocked(0)) > 0 {
 					assert.Equal(t,
-						pb.LockTable{ServiceID: "s1", Table: 1, Version: 1, Valid: false},
-						a.mu.lockTables[1])
+						pb.LockTable{ServiceID: "s1", Table: 1, Version: 1, OriginTable: 1, Valid: false},
+						a.getLockTablesLocked(0)[1])
 				}
 				a.mu.Unlock()
 				return
@@ -132,8 +132,10 @@ func TestKeepaliveBind(t *testing.T) {
 				assert.NoError(t, c.Close())
 			}()
 
-			bind := a.Get("s1", 1)
+			bind := a.Get("s1", 0, 1, 0, pb.Sharding_None)
+			gm := &sync.Map{}
 			m := &sync.Map{}
+			gm.Store(0, m)
 			m.Store(1,
 				newRemoteLockTable(
 					"s1",
@@ -141,7 +143,7 @@ func TestKeepaliveBind(t *testing.T) {
 					bind,
 					c,
 					func(lt pb.LockTable) {}))
-			k := NewLockTableKeeper("s1", c, interval/5, interval/5, m)
+			k := NewLockTableKeeper("s1", c, interval/5, interval/5, gm)
 
 			binds := a.getServiceBinds("s1")
 			assert.NotNil(t, binds)
@@ -154,7 +156,7 @@ func TestKeepaliveBind(t *testing.T) {
 
 			for {
 				a.mu.Lock()
-				valid := a.mu.lockTables[1].Valid
+				valid := a.getLockTablesLocked(0)[1].Valid
 				a.mu.Unlock()
 				if !valid {
 					break
@@ -171,7 +173,7 @@ func TestValid(t *testing.T) {
 		t,
 		time.Hour,
 		func(a *lockTableAllocator) {
-			b := a.Get("s1", 4)
+			b := a.Get("s1", 0, 4, 0, pb.Sharding_None)
 			assert.Empty(t, a.Valid([]pb.LockTable{b}))
 		})
 }
@@ -181,7 +183,7 @@ func TestValidWithServiceInvalid(t *testing.T) {
 		t,
 		time.Hour,
 		func(a *lockTableAllocator) {
-			b := a.Get("s1", 4)
+			b := a.Get("s1", 0, 4, 0, pb.Sharding_None)
 			b.ServiceID = "s2"
 			assert.NotEmpty(t, a.Valid([]pb.LockTable{b}))
 		})
@@ -192,7 +194,7 @@ func TestValidWithVersionChanged(t *testing.T) {
 		t,
 		time.Hour,
 		func(a *lockTableAllocator) {
-			b := a.Get("s1", 4)
+			b := a.Get("s1", 0, 4, 0, pb.Sharding_None)
 			b.Version++
 			assert.NotEmpty(t, a.Valid([]pb.LockTable{b}))
 		})
@@ -223,7 +225,7 @@ func runValidBenchmark(b *testing.B, name string, tables int) {
 		}()
 		var binds []pb.LockTable
 		for i := 0; i < tables; i++ {
-			binds = append(binds, a.Get(fmt.Sprintf("s-%d", i), uint64(i)))
+			binds = append(binds, a.Get(fmt.Sprintf("s-%d", i), 0, uint64(i), 0, pb.Sharding_None))
 		}
 		b.ReportAllocs()
 		b.ResetTimer()
diff --git a/pkg/lockservice/lock_table_keeper_test.go b/pkg/lockservice/lock_table_keeper_test.go
index 455c641a7..292e94758 100644
--- a/pkg/lockservice/lock_table_keeper_test.go
+++ b/pkg/lockservice/lock_table_keeper_test.go
@@ -62,7 +62,9 @@ func TestKeeper(t *testing.T) {
 					}
 					writeResponse(ctx, cancel, resp, nil, cs)
 				})
+			gm := &sync.Map{}
 			m := &sync.Map{}
+			gm.Store("", m)
 			m.Store(0,
 				newRemoteLockTable(
 					"s1",
@@ -82,7 +84,7 @@ func TestKeeper(t *testing.T) {
 				c,
 				time.Millisecond*10,
 				time.Millisecond*10,
-				m)
+				gm)
 			defer func() {
 				assert.NoError(t, k.Close())
 			}()
@@ -122,19 +124,23 @@ func TestKeepBindFailedWillRemoveAllLocalLockTable(t *testing.T) {
 					writeResponse(ctx, cancel, resp, nil, cs)
 				})
 
+			gm := &sync.Map{}
 			m := &sync.Map{}
+			gm.Store("", m)
 			m.Store(1,
 				newLocalLockTable(
 					pb.LockTable{ServiceID: "s1"},
 					nil,
 					events,
-					runtime.DefaultRuntime().Clock()))
+					runtime.DefaultRuntime().Clock(),
+					nil))
 			m.Store(2,
 				newLocalLockTable(
 					pb.LockTable{ServiceID: "s1"},
 					nil,
 					events,
-					runtime.DefaultRuntime().Clock()))
+					runtime.DefaultRuntime().Clock(),
+					nil))
 			m.Store(3,
 				newRemoteLockTable(
 					"s1",
@@ -147,7 +153,7 @@ func TestKeepBindFailedWillRemoveAllLocalLockTable(t *testing.T) {
 				c,
 				time.Millisecond*10,
 				time.Millisecond*10,
-				m)
+				gm)
 			defer func() {
 				assert.NoError(t, k.Close())
 			}()
diff --git a/pkg/lockservice/lock_table_local_test.go b/pkg/lockservice/lock_table_local_test.go
index b9aedb307..30aba977e 100644
--- a/pkg/lockservice/lock_table_local_test.go
+++ b/pkg/lockservice/lock_table_local_test.go
@@ -106,7 +106,7 @@ func TestCloseLocalLockTableWithBlockedWaiter(t *testing.T) {
 				require.Equal(t, ErrLockTableNotFound, err)
 			}()
 
-			v, err := l.getLockTable(tableID)
+			v, err := l.getLockTable(0, tableID)
 			require.NoError(t, err)
 			lt := v.(*localLockTable)
 			for {
@@ -347,7 +347,7 @@ func TestMergeRangeWithNoConflict(t *testing.T) {
 			table := uint64(10)
 			for _, c := range cases {
 				stopper := stopper.NewStopper("")
-				v, err := l.getLockTableWithCreate(table, true)
+				v, err := l.getLockTableWithCreate(0, table, nil, pb.Sharding_None)
 				require.NoError(t, err)
 				lt := v.(*localLockTable)
 
@@ -422,7 +422,7 @@ func TestMergeRangeWithNoConflict(t *testing.T) {
 					})
 					return values
 				}
-				assert.Equal(t, fn(c.mergedLocks), fn(txn.holdLocks[table].slice().all()))
+				assert.Equal(t, fn(c.mergedLocks), fn(txn.getHoldLocksLocked(0).tableKeys[table].slice().all()))
 
 				assert.NoError(t, l.Unlock(ctx, []byte(c.txnID), timestamp.Timestamp{}))
 				stopper.Stop()
@@ -481,7 +481,7 @@ func TestLocalLockTableMultipleRowLocksCannotMissIfFoundSelfTxn(t *testing.T) {
 			require.NoError(t, l.Unlock(ctx, []byte{2}, timestamp.Timestamp{}))
 
 			wg.Wait()
-			v, err := l.getLockTable(tableID)
+			v, err := l.getLockTable(0, tableID)
 			require.NoError(t, err)
 			lt := v.(*localLockTable)
 			lt.mu.Lock()
@@ -555,7 +555,7 @@ func TestIssue9856(t *testing.T) {
 				json.MustUnmarshal([]byte(r), v)
 				_, err := l.Lock(ctx, tableID, [][]byte{[]byte(v.Start), []byte(v.End)}, []byte("txn1"), option)
 				require.NoError(t, err)
-				vv, err := l.getLockTable(tableID)
+				vv, err := l.getLockTable(0, tableID)
 				require.NoError(t, err)
 				lt := vv.(*localLockTable)
 				lt.mu.Lock()
@@ -722,7 +722,7 @@ func TestLockedTSIsLastCommittedTS(t *testing.T) {
 			defer cancel()
 
 			tableID := uint64(10)
-			v, err := l.getLockTableWithCreate(tableID, true)
+			v, err := l.getLockTableWithCreate(0, tableID, nil, pb.Sharding_None)
 			require.NoError(t, err)
 			lt := v.(*localLockTable)
 			lt.mu.Lock()
@@ -793,7 +793,7 @@ func TestLockedTSIsLastCommittedTSWithRange(t *testing.T) {
 			defer cancel()
 
 			tableID := uint64(10)
-			v, err := l.getLockTableWithCreate(tableID, true)
+			v, err := l.getLockTableWithCreate(0, tableID, nil, pb.Sharding_None)
 			require.NoError(t, err)
 			lt := v.(*localLockTable)
 			lt.mu.Lock()
diff --git a/pkg/lockservice/lock_table_proxy_test.go b/pkg/lockservice/lock_table_proxy_test.go
new file mode 100644
index 000000000..c35481dda
--- /dev/null
+++ b/pkg/lockservice/lock_table_proxy_test.go
@@ -0,0 +1,146 @@
+// Copyright 2023 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lockservice
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProxySharedLock(t *testing.T) {
+	runLockServiceTests(
+		t,
+		[]string{"s1", "s2"},
+		func(alloc *lockTableAllocator, s []*service) {
+			tableID := uint64(10)
+			s1 := s[0]
+			s2 := s[1]
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second*1000000)
+			defer cancel()
+
+			option := newTestRowSharedOptions()
+			rows := newTestRows(1)
+			txn1 := newTestTxnID(1)
+			txn2 := newTestTxnID(2)
+			txn3 := newTestTxnID(3)
+			txn4 := newTestTxnID(4)
+			txn5 := newTestTxnID(5)
+
+			s1.cfg.EnableRemoteLocalProxy = true
+			_, err := s1.Lock(ctx, tableID, rows, txn1, option)
+			require.NoError(t, err, err)
+			require.NoError(t, s1.Unlock(ctx, txn1, timestamp.Timestamp{}))
+
+			v, _ := s1.getTables(0).Load(tableID)
+			lt := v.(*localLockTable)
+
+			// s2 will enable shared remote proxy
+			s2.cfg.EnableRemoteLocalProxy = true
+			_, err = s2.Lock(ctx, tableID, rows, txn2, option)
+			require.NoError(t, err)
+			checkLock(t, lt, rows[0], [][]byte{txn2}, nil, nil)
+			v, _ = s2.getTables(0).Load(tableID)
+			ltp := v.(*localLockTableProxy)
+			require.Equal(t, ltp.mu.currentHolder[string(rows[0])], txn2)
+			require.Equal(t, 1, len(ltp.mu.holders[string(rows[0])].txns))
+			require.Equal(t, ltp.mu.holders[string(rows[0])].txns[0].txnID, txn2)
+			require.True(t, ltp.mu.holders[string(rows[0])].cbs[0] == nil)
+			require.True(t, ltp.mu.holders[string(rows[0])].waiters[0] == nil)
+
+			_, err = s2.Lock(ctx, tableID, rows, txn3, option)
+			require.NoError(t, err)
+			checkLock(t, lt, rows[0], [][]byte{txn2}, nil, nil)
+			require.Equal(t, ltp.mu.currentHolder[string(rows[0])], txn2)
+			require.Equal(t, 2, len(ltp.mu.holders[string(rows[0])].txns))
+			require.Equal(t, ltp.mu.holders[string(rows[0])].txns[1].txnID, txn3)
+			require.True(t, ltp.mu.holders[string(rows[0])].cbs[1] == nil)
+			require.True(t, ltp.mu.holders[string(rows[0])].waiters[1] == nil)
+
+			_, err = s2.Lock(ctx, tableID, rows, txn4, option)
+			require.NoError(t, err)
+			checkLock(t, lt, rows[0], [][]byte{txn2}, nil, nil)
+			require.Equal(t, ltp.mu.currentHolder[string(rows[0])], txn2)
+			require.Equal(t, 3, len(ltp.mu.holders[string(rows[0])].txns))
+			require.Equal(t, ltp.mu.holders[string(rows[0])].txns[2].txnID, txn4)
+			require.True(t, ltp.mu.holders[string(rows[0])].cbs[2] == nil)
+			require.True(t, ltp.mu.holders[string(rows[0])].waiters[2] == nil)
+
+			require.NoError(t, s2.Unlock(ctx, txn2, timestamp.Timestamp{}))
+			checkLock(t, lt, rows[0], [][]byte{txn4}, nil, nil)
+			require.Equal(t, ltp.mu.currentHolder[string(rows[0])], txn4)
+
+			require.NoError(t, s2.Unlock(ctx, txn4, timestamp.Timestamp{}))
+			checkLock(t, lt, rows[0], [][]byte{txn3}, nil, nil)
+			require.Equal(t, ltp.mu.currentHolder[string(rows[0])], txn3)
+
+			require.NoError(t, s2.Unlock(ctx, txn3, timestamp.Timestamp{}))
+			checkLock(t, lt, rows[0], [][]byte{}, nil, nil)
+			require.Empty(t, ltp.mu.currentHolder)
+
+			_, err = s1.Lock(ctx, tableID, rows, txn5, newTestRowExclusiveOptions())
+			require.NoError(t, err, err)
+			require.NoError(t, s1.Unlock(ctx, txn5, timestamp.Timestamp{}))
+		},
+	)
+}
+
+func TestProxySharedUnlock(t *testing.T) {
+	runLockServiceTests(
+		t,
+		[]string{"s1", "s2"},
+		func(alloc *lockTableAllocator, s []*service) {
+			tableID := uint64(10)
+			s1 := s[0]
+			s2 := s[1]
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second*1000000)
+			defer cancel()
+
+			option := newTestRowSharedOptions()
+			rows := newTestRows(1)
+			txn1 := newTestTxnID(1)
+			txn2 := newTestTxnID(2)
+			txn3 := newTestTxnID(3)
+			txn4 := newTestTxnID(3)
+
+			s1.cfg.EnableRemoteLocalProxy = true
+			_, err := s1.Lock(ctx, tableID, rows, txn1, option)
+			require.NoError(t, err, err)
+			require.NoError(t, s1.Unlock(ctx, txn1, timestamp.Timestamp{}))
+
+			v, _ := s1.getTables(0).Load(tableID)
+			lt := v.(*localLockTable)
+
+			// s2 will enable shared remote proxy
+			s2.cfg.EnableRemoteLocalProxy = true
+			_, err = s2.Lock(ctx, tableID, rows, txn2, option)
+			require.NoError(t, err)
+
+			_, err = s2.Lock(ctx, tableID, rows, txn3, option)
+			require.NoError(t, err)
+			require.NoError(t, s2.Unlock(ctx, txn3, timestamp.Timestamp{}))
+
+			require.NoError(t, s2.Unlock(ctx, txn2, timestamp.Timestamp{}))
+			checkLock(t, lt, rows[0], [][]byte{}, nil, nil)
+
+			_, err = s1.Lock(ctx, tableID, rows, txn4, newTestRowExclusiveOptions())
+			require.NoError(t, err, err)
+			require.NoError(t, s1.Unlock(ctx, txn4, timestamp.Timestamp{}))
+		},
+	)
+}
diff --git a/pkg/lockservice/service_forward_test.go b/pkg/lockservice/service_forward_test.go
index 4b96d2270..2db5d7848 100644
--- a/pkg/lockservice/service_forward_test.go
+++ b/pkg/lockservice/service_forward_test.go
@@ -37,7 +37,7 @@ func TestForwardLock(t *testing.T) {
 			ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 			defer cancel()
 
-			_, err := l2.getLockTableWithCreate(tableID, true)
+			_, err := l2.getLockTableWithCreate(0, tableID, nil, pb.Sharding_None)
 			require.NoError(t, err)
 
 			txn1 := []byte("txn1")
diff --git a/pkg/lockservice/service_observability_test.go b/pkg/lockservice/service_observability_test.go
index 7a78576bd..b2b93744d 100644
--- a/pkg/lockservice/service_observability_test.go
+++ b/pkg/lockservice/service_observability_test.go
@@ -133,10 +133,10 @@ func TestGetLockTableBind(t *testing.T) {
 			alloc *lockTableAllocator,
 			l1, l2 *service,
 			table uint64) {
-			bind1, err := l1.GetLockTableBind(table)
+			bind1, err := l1.GetLockTableBind(0, table)
 			require.NoError(t, err)
 
-			bind2, err := l2.GetLockTableBind(table)
+			bind2, err := l2.GetLockTableBind(0, table)
 			require.NoError(t, err)
 
 			assert.Equal(t, bind1, bind2)
diff --git a/pkg/lockservice/service_remote_test.go b/pkg/lockservice/service_remote_test.go
index 0420e0d22..7810bd2e8 100644
--- a/pkg/lockservice/service_remote_test.go
+++ b/pkg/lockservice/service_remote_test.go
@@ -330,7 +330,7 @@ func TestLockWithBindIsStale(t *testing.T) {
 
 			checkBind(
 				t,
-				pb.LockTable{ServiceID: l1.serviceID, Version: 2, Table: table, Valid: true},
+				pb.LockTable{ServiceID: l1.serviceID, Version: 2, Table: table, OriginTable: table, Valid: true},
 				l2)
 		},
 	)
@@ -351,7 +351,7 @@ func TestUnlockWithBindIsStable(t *testing.T) {
 
 			checkBind(
 				t,
-				pb.LockTable{ServiceID: l1.serviceID, Version: 2, Table: table, Valid: true},
+				pb.LockTable{ServiceID: l1.serviceID, Version: 2, Table: table, OriginTable: table, Valid: true},
 				l2)
 		},
 	)
@@ -368,13 +368,13 @@ func TestGetLockWithBindIsStable(t *testing.T) {
 			table uint64) {
 
 			txnID2 := []byte("txn2")
-			lt, err := l2.getLockTable(table)
+			lt, err := l2.getLockTable(0, table)
 			require.NoError(t, err)
 			lt.getLock(txnID2, pb.WaitTxn{TxnID: []byte{1}}, func(l Lock) {})
 
 			checkBind(
 				t,
-				pb.LockTable{ServiceID: l1.serviceID, Version: 2, Table: table, Valid: true},
+				pb.LockTable{ServiceID: l1.serviceID, Version: 2, Table: table, OriginTable: table, Valid: true},
 				l2)
 		},
 	)
@@ -404,7 +404,7 @@ func TestLockWithBindTimeout(t *testing.T) {
 				})
 				if err == nil {
 					// l2 get the bind
-					v, ok := l2.tables.Load(table)
+					v, ok := l2.getTables(0).Load(table)
 					assert.True(t, ok)
 					l := v.(lockTable)
 					assert.Equal(t, l2.serviceID, l.getBind().ServiceID)
@@ -433,7 +433,7 @@ func TestUnlockWithBindTimeout(t *testing.T) {
 			txnID2 := []byte("txn2")
 			assert.NoError(t, l2.Unlock(ctx, txnID2, timestamp.Timestamp{}))
 			// l2 get the bind
-			v, ok := l2.tables.Load(table)
+			v, ok := l2.getTables(0).Load(table)
 			assert.True(t, ok)
 			l := v.(lockTable)
 			assert.Equal(t, l2.serviceID, l.getBind().ServiceID)
@@ -456,11 +456,11 @@ func TestGetLockWithBindTimeout(t *testing.T) {
 			waitBindDisabled(t, alloc, l1.serviceID)
 
 			txnID2 := []byte("txn2")
-			lt, err := l2.getLockTable(table)
+			lt, err := l2.getLockTable(0, table)
 			require.NoError(t, err)
 			lt.getLock(txnID2, pb.WaitTxn{TxnID: []byte{1}}, func(l Lock) {})
 			// l2 get the bind
-			v, ok := l2.tables.Load(table)
+			v, ok := l2.getTables(0).Load(table)
 			assert.True(t, ok)
 			l := v.(lockTable)
 			assert.Equal(t, l2.serviceID, l.getBind().ServiceID)
@@ -491,7 +491,7 @@ func TestLockWithBindNotFound(t *testing.T) {
 
 			checkBind(
 				t,
-				pb.LockTable{ServiceID: l1.serviceID, Version: 1, Table: table, Valid: true},
+				pb.LockTable{ServiceID: l1.serviceID, Version: 1, Table: table, OriginTable: table, Valid: true},
 				l2)
 		},
 	)
@@ -515,7 +515,7 @@ func TestUnlockWithBindNotFound(t *testing.T) {
 
 			checkBind(
 				t,
-				pb.LockTable{ServiceID: l1.serviceID, Version: 1, Table: table, Valid: true},
+				pb.LockTable{ServiceID: l1.serviceID, Version: 1, Table: table, OriginTable: table, Valid: true},
 				l2)
 		},
 	)
@@ -535,13 +535,13 @@ func TestGetLockWithBindNotFound(t *testing.T) {
 			l2.handleBindChanged(pb.LockTable{Table: table, ServiceID: "s3", Valid: true, Version: 1})
 
 			txnID2 := []byte("txn2")
-			lt, err := l2.getLockTable(table)
+			lt, err := l2.getLockTable(0, table)
 			require.NoError(t, err)
 			lt.getLock(txnID2, pb.WaitTxn{TxnID: []byte{1}}, func(l Lock) {})
 
 			checkBind(
 				t,
-				pb.LockTable{ServiceID: l1.serviceID, Version: 1, Table: table, Valid: true},
+				pb.LockTable{ServiceID: l1.serviceID, Version: 1, Table: table, OriginTable: table, Valid: true},
 				l2)
 		},
 	)
@@ -565,12 +565,12 @@ func TestIssue12554(t *testing.T) {
 			// txn1 hold lock row1 on l1
 			mustAddTestLock(t, ctx, l1, table, txn1, [][]byte{row1}, pb.Granularity_Row)
 
-			oldBind := alloc.Get(l1.serviceID, table)
+			oldBind := alloc.Get(l1.serviceID, 0, table, 0, pb.Sharding_None)
 			// mock l1 restart, changed serviceID
 			l1.serviceID = getServiceIdentifier("s1", time.Now().UnixNano())
-			l1.tables.Delete(table)
+			l1.getTables(0).Delete(table)
 			newLockTable := l1.createLockTableByBind(oldBind)
-			l1.tables.Store(table, newLockTable)
+			l1.getTables(0).Store(table, newLockTable)
 
 			_, err := l2.Lock(ctx, table, [][]byte{row1}, txn2, pb.LockOptions{
 				Granularity: pb.Granularity_Row,
@@ -609,14 +609,14 @@ func runBindChangedTests(
 
 			// l2 get the table1's bind
 			mustAddTestLock(t, ctx, l2, table1, txnID2, [][]byte{{2}}, pb.Granularity_Row)
-			v, err := l2.getLockTable(table1)
+			v, err := l2.getLockTable(0, table1)
 			require.NoError(t, err)
 			require.Equal(t, l1.serviceID, v.getBind().ServiceID)
 
 			if makeBindChanged {
 				// stop l1 keep lock bind
 				skip.Store(true)
-				lt, err := l1.getLockTable(table1)
+				lt, err := l1.getLockTable(0, table1)
 				require.NoError(t, err)
 				old := lt.getBind()
 				waitBindDisabled(t, alloc, l1.serviceID)
@@ -665,7 +665,7 @@ func waitBindChanged(
 	old pb.LockTable,
 	l *service) {
 	for {
-		lt, err := l.getLockTableWithCreate(old.Table, true)
+		lt, err := l.getLockTableWithCreate(0, old.Table, nil, pb.Sharding_None)
 		require.NoError(t, err)
 		new := lt.getBind()
 		if new.Changed(old) {
@@ -679,7 +679,7 @@ func checkBind(
 	t *testing.T,
 	bind pb.LockTable,
 	s *service) {
-	v, ok := s.tables.Load(bind.Table)
+	v, ok := s.getTables(0).Load(bind.Table)
 	assert.True(t, ok)
 	l := v.(lockTable)
 	assert.Equal(t, bind, l.getBind())
diff --git a/pkg/lockservice/service_test.go b/pkg/lockservice/service_test.go
index 0cfe82264..c71121bbd 100644
--- a/pkg/lockservice/service_test.go
+++ b/pkg/lockservice/service_test.go
@@ -16,6 +16,8 @@ package lockservice
 
 import (
 	"context"
+	"hash/crc32"
+	"hash/crc64"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -25,7 +27,6 @@ import (
 	"github.com/lni/goutils/leaktest"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/reuse"
-	"github.com/matrixorigin/matrixone/pkg/defines"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/lock"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
 	"github.com/stretchr/testify/assert"
@@ -60,8 +61,10 @@ func getRunner(remote bool) func(t *testing.T, table uint64, fn func(context.Con
 				require.NoError(t, err, err)
 				require.NoError(t, s1.Unlock(ctx, txn1, timestamp.Timestamp{}))
 
-				lt, err := s1.getLockTable(table)
+				lt, err := s1.getLockTable(0, table)
 				require.NoError(t, err)
+				require.Equal(t, table, lt.getBind().Table)
+				require.Equal(t, table, lt.getBind().OriginTable)
 
 				target := s1
 				if remote {
@@ -1075,7 +1078,7 @@ func TestLockResultWithNoConflict(t *testing.T) {
 			require.NoError(t, err)
 			assert.False(t, res.Timestamp.IsEmpty())
 
-			lb, err := l.getLockTable(0)
+			lb, err := l.getLockTable(0, 0)
 			require.NoError(t, err)
 			assert.Equal(t, lb.getBind(), res.LockedOn)
 		},
@@ -1348,8 +1351,8 @@ func TestTxnUnlockWithBindChanged(t *testing.T) {
 					// changed bind
 					bind := lt.getBind()
 					bind.Version = bind.Version + 1
-					new := newLocalLockTable(bind, s.fsp, s.events, s.clock)
-					s.tables.Store(table, new)
+					new := newLocalLockTable(bind, s.fsp, s.events, s.clock, s.activeTxnHolder)
+					s.getTables(0).Store(table, new)
 					lt.close()
 
 					// txn2 get lock, shared
@@ -1364,11 +1367,136 @@ func TestTxnUnlockWithBindChanged(t *testing.T) {
 	}
 }
 
+func TestMultiGroupWithSameTableID(t *testing.T) {
+	for name, runner := range runners {
+		t.Run(name, func(t *testing.T) {
+			table := uint64(10)
+			g1 := uint32(1)
+			g2 := uint32(2)
+			runner(
+				t,
+				table,
+				func(
+					ctx context.Context,
+					s *service,
+					lt *localLockTable) {
+
+					rows := newTestRows(1)
+
+					txn1 := newTestTxnID(1)
+					option1 := newTestRowSharedOptions()
+					option1.Group = g1
+
+					txn2 := newTestTxnID(2)
+					option2 := newTestRowSharedOptions()
+					option2.Group = g2
+
+					// txn1 get lock
+					_, err := s.Lock(ctx, table, rows, txn1, option1)
+					require.NoError(t, err)
+					lt1, err := s.getLockTable(g1, table)
+					assert.NoError(t, err)
+					checkLock(t, lt1.(*localLockTable), rows[0], [][]byte{txn1}, nil, nil)
+
+					// txn2 get lock, shared
+					_, err = s.Lock(ctx, table, rows, txn2, option2)
+					require.NoError(t, err)
+					lt2, err := s.getLockTable(g2, table)
+					assert.NoError(t, err)
+					checkLock(t, lt2.(*localLockTable), rows[0], [][]byte{txn2}, nil, nil)
+
+					require.NoError(t, s.Unlock(ctx, txn1, timestamp.Timestamp{}))
+					require.NoError(t, s.Unlock(ctx, txn2, timestamp.Timestamp{}))
+				})
+		})
+	}
+}
+
+func TestShardingByRowWithSameTableID(t *testing.T) {
+	for name, runner := range runners {
+		t.Run(name, func(t *testing.T) {
+			table := uint64(10)
+			runner(
+				t,
+				table,
+				func(
+					ctx context.Context,
+					s *service,
+					lt *localLockTable) {
+
+					option := newTestRowSharedOptions()
+					option.Sharding = pb.Sharding_ByRow
+					option.Group = 1
+					txn1 := newTestTxnID(1)
+					txn2 := newTestTxnID(2)
+					rows1 := newTestRows(1)
+					rows2 := newTestRows(2)
+
+					// txn1 get lock
+					_, err := s.Lock(ctx, table, rows1, txn1, option)
+					require.NoError(t, err)
+					l := s.loadLockTable(1, shardingByRow(rows1[0]))
+					assert.Equal(t, table, l.getBind().OriginTable)
+					checkLock(t, l.(*localLockTable), rows1[0], [][]byte{txn1}, nil, nil)
+
+					// txn2 get lock, shared
+					_, err = s.Lock(ctx, table, rows2, txn2, option)
+					require.NoError(t, err)
+					l = s.loadLockTable(1, shardingByRow(rows2[0]))
+					assert.Equal(t, table, l.getBind().OriginTable)
+					checkLock(t, l.(*localLockTable), rows2[0], [][]byte{txn2}, nil, nil)
+
+					require.NoError(t, s.Unlock(ctx, txn1, timestamp.Timestamp{}))
+					require.NoError(t, s.Unlock(ctx, txn2, timestamp.Timestamp{}))
+				})
+		})
+	}
+}
+
 func BenchmarkWithoutConflict(b *testing.B) {
 	runBenchmark(b, "1-table", 1)
 	runBenchmark(b, "unlimited-table", 32)
 }
 
+func BenchmarkCrc64(b *testing.B) {
+	b.Run("crc64-ISO", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		b.RunParallel(func(p *testing.PB) {
+			sum := uint64(0)
+			for p.Next() {
+				n := crc64.Checksum([]byte("hello"), crc64.MakeTable(crc64.ISO))
+				sum += n
+			}
+			b.Log(sum)
+		})
+	})
+	b.Run("crc64-ECMA", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		b.RunParallel(func(p *testing.PB) {
+			sum := uint64(0)
+			for p.Next() {
+				n := crc64.Checksum([]byte("hello"), crc64.MakeTable(crc64.ECMA))
+				sum += n
+			}
+			b.Log(sum)
+		})
+	})
+	b.Run("crc32", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+		b.RunParallel(func(p *testing.PB) {
+			sum := uint32(0)
+			for p.Next() {
+				n := crc32.Checksum([]byte("hello"), crc32.MakeTable(crc32.IEEE))
+				sum += n
+			}
+			b.Log(sum)
+		})
+	})
+}
+
 var tableID atomic.Uint64
 var txnID atomic.Uint64
 var rowID atomic.Uint64
@@ -1499,7 +1627,7 @@ func waitWaiters(
 	table uint64,
 	key []byte,
 	waitersCount int) {
-	require.NoError(t, WaitWaiters(s, table, key, waitersCount))
+	require.NoError(t, WaitWaiters(s, 0, table, key, waitersCount))
 }
 
 func newTestRowExclusiveOptions() pb.LockOptions {
@@ -1601,14 +1729,3 @@ func mustAddTestLock(t *testing.T,
 		lock,
 		granularity)
 }
-
-func TestSharedTableID(t *testing.T) {
-	tenantID := uint32(955)
-	tableID := uint64(2)
-
-	ctx := context.WithValue(context.Background(), defines.TenantIDKey{}, tenantID)
-	tenantID2, tableID2, ok := decodeSharedTableID(encodeSharedTableID(ctx, tableID))
-	require.True(t, ok)
-	require.Equal(t, tenantID, tenantID2)
-	require.Equal(t, tableID, tableID2)
-}
diff --git a/pkg/lockservice/test_helper.go b/pkg/lockservice/test_helper.go
index c23c4eeca..f6407c735 100644
--- a/pkg/lockservice/test_helper.go
+++ b/pkg/lockservice/test_helper.go
@@ -89,11 +89,12 @@ func RunLockServicesForTest(
 // WaitWaiters wait waiters
 func WaitWaiters(
 	ls LockService,
+	group uint32,
 	table uint64,
 	key []byte,
 	waitersCount int) error {
 	s := ls.(*service)
-	v, err := s.getLockTable(table)
+	v, err := s.getLockTable(group, table)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/lockservice/txn_test.go b/pkg/lockservice/txn_test.go
index b55dccdb5..4adf5d5e8 100644
--- a/pkg/lockservice/txn_test.go
+++ b/pkg/lockservice/txn_test.go
@@ -33,18 +33,18 @@ func TestLockAdded(t *testing.T) {
 		txn := newActiveTxn(id, string(id), fsp, "")
 		defer reuse.Free(txn, nil)
 
-		txn.lockAdded(pb.LockTable{Table: 1}, [][]byte{[]byte("k1")})
-		txn.lockAdded(pb.LockTable{Table: 1}, [][]byte{[]byte("k11")})
-		txn.lockAdded(pb.LockTable{Table: 2}, [][]byte{[]byte("k2"), []byte("k22")})
+		txn.lockAdded(0, pb.LockTable{Table: 1}, [][]byte{[]byte("k1")})
+		txn.lockAdded(0, pb.LockTable{Table: 1}, [][]byte{[]byte("k11")})
+		txn.lockAdded(0, pb.LockTable{Table: 2}, [][]byte{[]byte("k2"), []byte("k22")})
 
-		assert.Equal(t, 2, len(txn.holdLocks))
+		assert.Equal(t, 2, len(txn.getHoldLocksLocked(0).tableKeys))
 
-		sp := txn.holdLocks[1]
+		sp := txn.getHoldLocksLocked(0).tableKeys[1]
 		s := sp.slice()
 		defer s.unref()
 		assert.Equal(t, 2, s.len())
 
-		sp2 := txn.holdLocks[2]
+		sp2 := txn.getHoldLocksLocked(0).tableKeys[2]
 		s2 := sp2.slice()
 		defer s2.unref()
 		assert.Equal(t, 2, s2.len())
@@ -60,8 +60,8 @@ func TestClose(t *testing.T) {
 		fsp := newFixedSlicePool(2)
 		txn := newActiveTxn(id, string(id), fsp, "")
 		tables := map[uint64]lockTable{
-			1: newLocalLockTable(pb.LockTable{Table: 1}, nil, events, runtime.DefaultRuntime().Clock()),
-			2: newLocalLockTable(pb.LockTable{Table: 2}, nil, events, runtime.DefaultRuntime().Clock()),
+			1: newLocalLockTable(pb.LockTable{Table: 1}, nil, events, runtime.DefaultRuntime().Clock(), nil),
+			2: newLocalLockTable(pb.LockTable{Table: 2}, nil, events, runtime.DefaultRuntime().Clock(), nil),
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
 		defer cancel()
@@ -78,13 +78,14 @@ func TestClose(t *testing.T) {
 			"s1",
 			txn.txnID,
 			timestamp.Timestamp{},
-			func(table uint64) (lockTable, error) {
+			func(group uint32, table uint64) (lockTable, error) {
 				return tables[table], nil
 			})
 		assert.Empty(t, txn.txnID)
 		assert.Empty(t, txn.txnKey)
 		assert.Empty(t, txn.blockedWaiters)
-		assert.Empty(t, txn.holdLocks)
+		assert.Empty(t, txn.getHoldLocksLocked(0).tableKeys)
+		assert.Empty(t, txn.getHoldLocksLocked(0).tableBinds)
 		assert.Equal(t, 0, tables[1].(*localLockTable).mu.store.Len())
 		assert.Equal(t, 0, tables[2].(*localLockTable).mu.store.Len())
 	})
diff --git a/pkg/sql/colexec/lockop/lock_op_test.go b/pkg/sql/colexec/lockop/lock_op_test.go
index 5867fd544..8bc494bf7 100644
--- a/pkg/sql/colexec/lockop/lock_op_test.go
+++ b/pkg/sql/colexec/lockop/lock_op_test.go
@@ -108,7 +108,7 @@ func TestCallLockOpWithConflict(t *testing.T) {
 					assert.Equal(t, types.BuildTS(math.MaxInt64, 1), v)
 				}
 			}()
-			require.NoError(t, lockservice.WaitWaiters(proc.LockService, tableID, conflictRow, 1))
+			require.NoError(t, lockservice.WaitWaiters(proc.LockService, 0, tableID, conflictRow, 1))
 			require.NoError(t, proc.LockService.Unlock(proc.Ctx, []byte("txn01"), timestamp.Timestamp{PhysicalTime: math.MaxInt64}))
 			<-c
 		},
@@ -164,7 +164,7 @@ func TestCallLockOpWithConflictWithRefreshNotEnabled(t *testing.T) {
 				require.Error(t, err)
 				assert.True(t, moerr.IsMoErrCode(err, moerr.ErrTxnNeedRetry))
 			}()
-			require.NoError(t, lockservice.WaitWaiters(proc.LockService, tableID, conflictRow, 1))
+			require.NoError(t, lockservice.WaitWaiters(proc.LockService, 0, tableID, conflictRow, 1))
 			require.NoError(t, proc.LockService.Unlock(proc.Ctx, []byte("txn01"), timestamp.Timestamp{PhysicalTime: math.MaxInt64}))
 			<-c
 		},
@@ -230,7 +230,7 @@ func TestCallLockOpWithHasPrevCommit(t *testing.T) {
 				require.Error(t, err)
 				assert.True(t, moerr.IsMoErrCode(err, moerr.ErrTxnNeedRetry))
 			}()
-			require.NoError(t, lockservice.WaitWaiters(proc.LockService, tableID, conflictRow, 1))
+			require.NoError(t, lockservice.WaitWaiters(proc.LockService, 0, tableID, conflictRow, 1))
 			require.NoError(t, proc.LockService.Unlock(proc.Ctx, []byte("txn02"), timestamp.Timestamp{}))
 			<-c
 		},
@@ -297,7 +297,7 @@ func TestCallLockOpWithHasPrevCommitLessMe(t *testing.T) {
 				_, err = arg2.Call(proc)
 				require.NoError(t, err)
 			}()
-			require.NoError(t, lockservice.WaitWaiters(proc.LockService, tableID, conflictRow, 1))
+			require.NoError(t, lockservice.WaitWaiters(proc.LockService, 0, tableID, conflictRow, 1))
 			require.NoError(t, proc.LockService.Unlock(proc.Ctx, []byte("txn02"), timestamp.Timestamp{}))
 			<-c
 		},
@@ -365,7 +365,7 @@ func TestLockWithBlockingWithConflict(t *testing.T) {
 			require.NoError(t, err)
 
 			go func() {
-				require.NoError(t, lockservice.WaitWaiters(proc.LockService, tableID, conflictRow, 1))
+				require.NoError(t, lockservice.WaitWaiters(proc.LockService, 0, tableID, conflictRow, 1))
 				require.NoError(t, proc.LockService.Unlock(
 					proc.Ctx,
 					[]byte("txn01"),
diff --git a/pkg/txn/client/operator_test.go b/pkg/txn/client/operator_test.go
index 6e8a7697c..79250c893 100644
--- a/pkg/txn/client/operator_test.go
+++ b/pkg/txn/client/operator_test.go
@@ -200,12 +200,12 @@ func TestCommitWithLockTablesChanged(t *testing.T) {
 				assert.Error(t, err)
 
 				// table 1 will be removed
-				bind, err := s.GetLockTableBind(tableID1)
+				bind, err := s.GetLockTableBind(0, tableID1)
 				require.NoError(t, err)
 				require.Equal(t, lock.LockTable{}, bind)
 
 				// table 2 will be kept
-				bind, err = s.GetLockTableBind(tableID2)
+				bind, err = s.GetLockTableBind(0, tableID2)
 				require.NoError(t, err)
 				require.NotEqual(t, lock.LockTable{}, bind)
 			},
diff --git a/pkg/txn/rpc/server_test.go b/pkg/txn/rpc/server_test.go
index a16e7b2df..df66776e9 100644
--- a/pkg/txn/rpc/server_test.go
+++ b/pkg/txn/rpc/server_test.go
@@ -217,6 +217,10 @@ func (cs *testClientSession) Write(ctx context.Context, response morpc.Message)
 	return nil
 }
 
+func (cs *testClientSession) AsyncWrite(response morpc.Message) error {
+	return nil
+}
+
 func (cs *testClientSession) CreateCache(
 	ctx context.Context,
 	cacheID uint64) (morpc.MessageCache, error) {
diff --git a/pkg/txn/service/service_cn_handler_test.go b/pkg/txn/service/service_cn_handler_test.go
index 6de279b74..897224c3a 100644
--- a/pkg/txn/service/service_cn_handler_test.go
+++ b/pkg/txn/service/service_cn_handler_test.go
@@ -27,6 +27,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/lockservice"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/pb/lock"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
 	"github.com/matrixorigin/matrixone/pkg/pb/txn"
@@ -642,7 +643,7 @@ func TestCommitWithLockTables(t *testing.T) {
 	}()
 	sender.AddTxnService(s)
 
-	bind := allocator.Get("s1", 10)
+	bind := allocator.Get("s1", 0, 10, 0, lock.Sharding_None)
 	wTxn := NewTestTxn(1, 1, 1)
 	wTxn.LockTables = append(wTxn.LockTables, bind)
 	checkResponses(t, writeTestData(t, sender, 1, wTxn, 0))
@@ -685,7 +686,7 @@ func TestCommitWithLockTablesBindChanged(t *testing.T) {
 	}()
 	sender.AddTxnService(s)
 
-	bind := allocator.Get("s1", 10)
+	bind := allocator.Get("s1", 0, 10, 0, lock.Sharding_None)
 	wTxn := NewTestTxn(1, 1, 1)
 	bind.ServiceID = "s2"
 	wTxn.LockTables = append(wTxn.LockTables, bind)
diff --git a/pkg/vm/engine/tae/logtail/service/session_test.go b/pkg/vm/engine/tae/logtail/service/session_test.go
index 507f5096e..52fc6287c 100644
--- a/pkg/vm/engine/tae/logtail/service/session_test.go
+++ b/pkg/vm/engine/tae/logtail/service/session_test.go
@@ -307,6 +307,11 @@ func (m *blockStream) Write(ctx context.Context, message morpc.Message) error {
 	return moerr.NewStreamClosedNoCtx()
 }
 
+func (m *blockStream) AsyncWrite(message morpc.Message) error {
+	<-m.ch
+	return moerr.NewStreamClosedNoCtx()
+}
+
 func (m *blockStream) Close() error {
 	m.once.Do(func() {
 		close(m.ch)
@@ -342,6 +347,10 @@ func (m *brokenStream) Write(ctx context.Context, message morpc.Message) error {
 	return moerr.NewStreamClosedNoCtx()
 }
 
+func (cs *brokenStream) AsyncWrite(response morpc.Message) error {
+	return nil
+}
+
 func (m *brokenStream) Close() error {
 	return nil
 }
@@ -380,6 +389,12 @@ func (m *normalStream) Write(ctx context.Context, message morpc.Message) error {
 	return nil
 }
 
+func (m *normalStream) AsyncWrite(message morpc.Message) error {
+	response := message.(*LogtailResponseSegment)
+	m.logger.Info("write response segment:", zap.String("segment", response.String()))
+	return nil
+}
+
 func (m *normalStream) Close() error {
 	return nil
 }