func (rs *RPCServer) Commit(ctx context.Context, req *tspb.CommitRequest) (*tspb.CommitResponse, error) {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		commitCounterGerneralErr.Inc()
		return nil, err
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	txnSel := &tspb.TransactionSelector{}
	switch req.Transaction.(type) {
	case *tspb.CommitRequest_SingleUseTransaction:
		txnSel.Selector = &tspb.TransactionSelector_SingleUse{
			SingleUse: req.GetSingleUseTransaction(),
		}
	case *tspb.CommitRequest_TransactionId:
		txnSel.Selector = &tspb.TransactionSelector_Id{
			Id: req.GetTransactionId(),
		}
	default:
		return nil, status.Errorf(codes.Aborted, "unsupported transaction type %T", req.Transaction)
	}
	if len(req.Table) == 0 {
		return nil, status.Error(codes.InvalidArgument, "missing field <table>")
	}
	isRawkv, err := sess.RawkvAccess(ctx, req.Table)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: true,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, txnSel, &rtOpt)
	if err != nil {
		commitCounterGerneralErr.Inc()
		return nil, status.Errorf(codes.Aborted, "retrieve session transaction failed %v", err)
	}
	startTS := time.Now()
	r, err := sess.HandleMutate(ctx, req, txn)
	durMutate := time.Since(startTS)
	if err != nil {
		executeMutateDurationGeneralErr.Observe(durMutate.Seconds())
		commitCounterGerneralErr.Inc()
		return nil, err
	}
	executeMutateDurationGeneralOK.Observe(durMutate.Seconds())
	commitCounterGerneralOK.Inc()
	return rpc.BuildCommitResponse(r), nil
}
Commit commits a transaction. The request includes the mutations to be
applied to rows in the database.

`Commit` might return an `ABORTED` error. This can occur at any time;
commonly, the cause is conflicts with concurrent
transactions. However, it can also happen for a variety of other
reasons. If `Commit` returns `ABORTED`, the caller should re-attempt
the transaction from the beginning, re-using the same session.
package server
import (
	"context"
	"fmt"
	"io"
	"sync"
	"sync/atomic"
	"time"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	emptypb "github.com/gogo/protobuf/types"
	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
	grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/pingcap/tidb/util/logutil"
	tspb "github.com/zhihu/zetta-proto/pkg/tablestore"
	"github.com/zhihu/zetta/pkg/metrics"
	"github.com/zhihu/zetta/tablestore/domain"
	"github.com/zhihu/zetta/tablestore/rpc"
	"github.com/zhihu/zetta/tablestore/session"
	"github.com/zhihu/zetta/tablestore/sessionctx"
)
var (
	grpcMetrics = grpc_prometheus.NewServerMetrics()
	createSessionGerneralOK = metrics.CreateSessionCounter.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	deleteSessionGerneralOK = metrics.DeleteSessionCounter.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	readCounterGerneralOK        = metrics.ReadCounter.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	readCounterGerneralErr       = metrics.ReadCounter.WithLabelValues(metrics.LblGeneral, metrics.LblError)
	streamReadCounterGerneralOK  = metrics.StreamReadCounter.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	streamReadCounterGerneralErr = metrics.StreamReadCounter.WithLabelValues(metrics.LblGeneral, metrics.LblError)
	sparseReadCounterGerneralOK  = metrics.SparseReadCounter.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	sparseReadCounterGerneralErr = metrics.SparseReadCounter.WithLabelValues(metrics.LblGeneral, metrics.LblError)
	commitCounterGerneralOK  = metrics.CommitCounter.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	commitCounterGerneralErr = metrics.CommitCounter.WithLabelValues(metrics.LblGeneral, metrics.LblError)
	mutateCounterGerneralOK  = metrics.MutateCounter.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	mutateCounterGerneralErr = metrics.MutateCounter.WithLabelValues(metrics.LblGeneral, metrics.LblError)
	executeMutateDurationGeneralOK  = metrics.ExecuteMutateDuration.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	executeMutateDurationGeneralErr = metrics.ExecuteMutateDuration.WithLabelValues(metrics.LblGeneral, metrics.LblError)
	executeReadDurationGeneralOK    = metrics.ExecuteReadDuration.WithLabelValues(metrics.LblGeneral, metrics.LblOK)
	executeReadDurationGeneralErr   = metrics.ExecuteReadDuration.WithLabelValues(metrics.LblGeneral, metrics.LblError)
)
type RPCServer struct {
	s   *Server
	rpc *grpc.Server
	mu  sync.Mutex
	tspb.TablestoreAdminServer
	tspb.TablestoreServer
}
func NewRPCServer(s *Server) *RPCServer {
	rs := &RPCServer{s: s}
	unaryInterceptor := grpc_middleware.ChainUnaryServer(
		grpc_prometheus.UnaryServerInterceptor,
		grpc_opentracing.UnaryServerInterceptor(),
	)
	streamInterceptor := grpc_middleware.ChainStreamServer(
		grpc_prometheus.StreamServerInterceptor,
		grpc_opentracing.StreamServerInterceptor(),
	)
	rs.rpc = grpc.NewServer(
		grpc.StreamInterceptor(streamInterceptor),
		grpc.UnaryInterceptor(unaryInterceptor),
	)
	tspb.RegisterTablestoreAdminServer(rs.rpc, rs)
	tspb.RegisterTablestoreServer(rs.rpc, rs)
	return rs
}
func (rs *RPCServer) Run() {
	go rs.rpc.Serve(rs.s.listener)
}
func (rs *RPCServer) Close() {
	rs.mu.Lock()
	defer rs.mu.Unlock()
	rs.rpc.GracefulStop()
}
func (rs *RPCServer) CreateSession(ctx context.Context, req *tspb.CreateSessionRequest) (*tspb.Session, error) {
	if req.Database == "" {
		return nil, status.Errorf(codes.InvalidArgument, "database null")
	}
	is := domain.GetOnlyDomain().InfoSchema()
	if _, ok := is.GetDatabaseMetaByName(req.Database); !ok {
		return nil, status.Errorf(codes.NotFound, "no such database %s", req.Database)
	}
	queryID := atomic.AddUint32(&baseQueryID, 1)
	rs.s.rwlock.Lock()
	defer rs.s.rwlock.Unlock()
	if len(rs.s.queryCtxs)+1 > rs.s.cfg.MaxSessions {
		return nil, status.Errorf(codes.ResourceExhausted, "session count exceed max-sesions %v", rs.s.cfg.MaxSessions)
	}
	qctx, err := rs.s.driver.OpenCtx(uint64(queryID), req.Database)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "open query ctx error %v", err)
	}
	if req.Session != nil {
		qctx.GetSession().SetLabels(req.Session.Labels)
	}
	rs.s.queryCtxs[qctx.GetSession().GetName()] = qctx
	createSessionGerneralOK.Add(1)
	return qctx.GetSession().ToProto(), nil
}
func (rs *RPCServer) BatchCreateSessions(ctx context.Context, req *tspb.BatchCreateSessionsRequest) (*tspb.BatchCreateSessionsResponse, error) {
	if req.Database == "" {
		return nil, status.Errorf(codes.InvalidArgument, "database null")
	}
	is := domain.GetOnlyDomain().InfoSchema()
	if _, ok := is.GetDatabaseMetaByName(req.Database); !ok {
		return nil, status.Errorf(codes.NotFound, "no such database %s", req.Database)
	}
	if req.SessionTemplate == nil {
		return nil, status.Errorf(codes.InvalidArgument, "session template is required")
	}
	var sessions []*tspb.Session
	rs.s.rwlock.Lock()
	defer rs.s.rwlock.Unlock()
	if len(rs.s.queryCtxs)+int(req.GetSessionCount()) > rs.s.cfg.MaxSessions {
		return nil, status.Errorf(codes.ResourceExhausted, "session count exceed max-sesions %v", rs.s.cfg.MaxSessions)
	}
	for i := int32(0); i < req.GetSessionCount(); i++ {
		queryID := atomic.AddUint32(&baseQueryID, 1)
		qctx, err := rs.s.driver.OpenCtx(uint64(queryID), req.Database)
		if err != nil {
			return nil, status.Errorf(codes.Aborted, "create session error %v", err)
		}
		qctx.GetSession().SetLabels(req.SessionTemplate.Labels)
		rs.s.queryCtxs[qctx.GetSession().GetName()] = qctx
		sessions = append(sessions, qctx.GetSession().ToProto())
		createSessionGerneralOK.Add(1)
	}
	return &tspb.BatchCreateSessionsResponse{Session: sessions}, nil
}
func (rs *RPCServer) GetSession(ctx context.Context, req *tspb.GetSessionRequest) (*tspb.Session, error) {
	rs.s.rwlock.RLock()
	defer rs.s.rwlock.RUnlock()
	queryCtx, ok := rs.s.queryCtxs[req.Name]
	if !ok {
		return nil, status.Errorf(codes.NotFound, "session %s not found", req.Name)
	}
	session := queryCtx.GetSession()
	session.SetLastActive(time.Now())
	return session.ToProto(), nil
}
func (rs *RPCServer) ListSessions(ctx context.Context, req *tspb.ListSessionsRequest) (*tspb.ListSessionsResponse, error) {
	rs.s.rwlock.RLock()
	defer rs.s.rwlock.RUnlock()
	sessions := []*tspb.Session{}
	for _, qctx := range rs.s.queryCtxs {
		sess := qctx.GetSession().ToProto()
		sessions = append(sessions, sess)
	}
	rsp := &tspb.ListSessionsResponse{
		Sessions: sessions,
	}
	return rsp, nil
}
func (rs *RPCServer) DeleteSession(ctx context.Context, req *tspb.DeleteSessionRequest) (*emptypb.Empty, error) {
	rs.s.rwlock.Lock()
	defer rs.s.rwlock.Unlock()
	queryCtx, ok := rs.s.queryCtxs[req.Name]
	if !ok {
		return &emptypb.Empty{}, nil
	}
	if err := queryCtx.Close(); err != nil {
		logutil.Logger(ctx).Error("session close error", zap.Error(err), zap.String("session", req.Name))
		return nil, status.Errorf(codes.Aborted, fmt.Sprintf("session %s close error, try later", req.Name))
	}
	delete(rs.s.queryCtxs, req.Name)
	deleteSessionGerneralOK.Add(1)
	return &emptypb.Empty{}, nil
}
func (rs *RPCServer) Read(ctx context.Context, req *tspb.ReadRequest) (*tspb.ResultSet, error) {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		readCounterGerneralErr.Inc()
		return nil, err
	}
	if len(req.ResumeToken) > 0 {
		return nil, fmt.Errorf("read resumption not supported")
	}
	if len(req.PartitionToken) > 0 {
		return nil, fmt.Errorf("partition restrictions not supported")
	}
	if req.GetTransaction() == nil {
		readCounterGerneralErr.Inc()
		return nil, status.Error(codes.FailedPrecondition, "no transaction selector in grpc request")
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	isRawkv, err := sess.RawkvAccess(ctx, req.Table)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, req.GetTransaction(), &rtOpt)
	if err != nil {
		readCounterGerneralErr.Inc()
		logutil.Logger(ctx).Error("fetch transaction error", zap.Error(err))
		return nil, err
	}
	startTS := time.Now()
	ri, err := queryCtx.GetSession().HandleRead(ctx, req, txn)
	if err != nil {
		readCounterGerneralErr.Inc()
		logutil.Logger(ctx).Error("read error", zap.Error(err))
		return nil, err
	}
	resultSet, err := rs.readRows(ctx, ri)
	durRead := time.Since(startTS)
	if err != nil {
		executeReadDurationGeneralErr.Observe(durRead.Seconds())
		sparseReadCounterGerneralErr.Inc()
		return nil, err
	}
	executeReadDurationGeneralOK.Observe(durRead.Seconds())
	readCounterGerneralOK.Inc()
	return resultSet, nil
}
func (rs *RPCServer) readRows(ctx context.Context, ri session.RecordSet) (*tspb.ResultSet, error) {
	resultSet := &tspb.ResultSet{
		Metadata: buildResultSetMetaData(ri.Columns()),
		Rows:     []*tspb.ListValue{},
	}
	for {
		row, err := ri.Next(ctx)
		if err != nil {
			break
		}
		resultSet.SliceRows = append(resultSet.SliceRows, row.(*tspb.SliceCell))
	}
	if ri.LastErr() != nil {
		return nil, ri.LastErr()
	}
	return resultSet, nil
}
func (rs *RPCServer) SparseRead(ctx context.Context, req *tspb.SparseReadRequest) (*tspb.ResultSet, error) {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		sparseReadCounterGerneralErr.Inc()
		return nil, err
	}
	if len(req.ResumeToken) > 0 {
		return nil, fmt.Errorf("read resumption not supported")
	}
	if len(req.PartitionToken) > 0 {
		return nil, fmt.Errorf("partition restrictions not supported")
	}
	if req.GetTransaction() == nil {
		sparseReadCounterGerneralErr.Inc()
		return nil, status.Error(codes.FailedPrecondition, "no transaction selector in grpc request")
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	isRawkv, err := sess.RawkvAccess(ctx, req.Table)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, req.GetTransaction(), &rtOpt)
	if err != nil {
		sparseReadCounterGerneralErr.Inc()
		logutil.Logger(ctx).Error("fetch transaction error", zap.Error(err))
		return nil, err
	}
	startTS := time.Now()
	ri, err := queryCtx.GetSession().HandleRead(ctx, req, txn)
	if err != nil {
		sparseReadCounterGerneralErr.Inc()
		logutil.Logger(ctx).Error("read error", zap.Error(err))
		return nil, err
	}
	resultSet, err := rs.readRows(ctx, ri)
	durRead := time.Since(startTS)
	if err != nil {
		executeReadDurationGeneralErr.Observe(durRead.Seconds())
		sparseReadCounterGerneralErr.Inc()
		return resultSet, err
	}
	executeReadDurationGeneralOK.Observe(durRead.Seconds())
	sparseReadCounterGerneralOK.Inc()
	return resultSet, nil
}
func (rs *RPCServer) SparseScan(ctx context.Context, req *tspb.SparseScanRequest) (*tspb.ResultSet, error) {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		sparseReadCounterGerneralErr.Inc()
		return nil, err
	}
	if len(req.ResumeToken) > 0 {
		return nil, fmt.Errorf("read resumption not supported")
	}
	if len(req.PartitionToken) > 0 {
		return nil, fmt.Errorf("partition restrictions not supported")
	}
	if req.GetTransaction() == nil {
		sparseReadCounterGerneralErr.Inc()
		return nil, status.Error(codes.FailedPrecondition, "no transaction selector in grpc request")
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	isRawkv, err := sess.RawkvAccess(ctx, req.Table)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, req.GetTransaction(), &rtOpt)
	if err != nil {
		sparseReadCounterGerneralErr.Inc()
		logutil.Logger(ctx).Error("fetch transaction error", zap.Error(err))
		return nil, err
	}
	startTS := time.Now()
	ri, err := queryCtx.GetSession().HandleRead(ctx, req, txn)
	if err != nil {
		sparseReadCounterGerneralErr.Inc()
		logutil.Logger(ctx).Error("read error", zap.Error(err))
		return nil, err
	}
	resultSet, err := rs.readRows(ctx, ri)
	durRead := time.Since(startTS)
	if err != nil {
		executeReadDurationGeneralErr.Observe(durRead.Seconds())
		sparseReadCounterGerneralErr.Inc()
		return resultSet, err
	}
	executeReadDurationGeneralOK.Observe(durRead.Seconds())
	sparseReadCounterGerneralOK.Inc()
	return resultSet, nil
}
func (rs *RPCServer) Mutate(ctx context.Context, req *tspb.MutationRequest) (*tspb.MutationResponse, error) {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		mutateCounterGerneralErr.Inc()
		return nil, err
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	if req.GetTransaction() == nil {
		return nil, status.Errorf(codes.FailedPrecondition, "transaction selector should be specific")
	}
	isRawkv, err := sess.RawkvAccess(ctx, req.Table)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: true,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, req.GetTransaction(), &rtOpt)
	if err != nil {
		mutateCounterGerneralErr.Inc()
		return nil, status.Errorf(codes.Aborted, "retrieve session transaction failed %v", err)
	}
	startTS := time.Now()
	r, err := sess.HandleMutate(ctx, req, txn)
	durMutate := time.Since(startTS)
	if err != nil {
		executeMutateDurationGeneralErr.Observe(durMutate.Seconds())
		mutateCounterGerneralErr.Inc()
		return nil, err
	}
	executeMutateDurationGeneralOK.Observe(durMutate.Seconds())
	mutateCounterGerneralOK.Inc()
	return rpc.BuildMutationResponse(r), nil
}
func (rs *RPCServer) StreamingRead(req *tspb.ReadRequest, stream tspb.Tablestore_StreamingReadServer) error {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		return err
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	isRawkv, err := sess.RawkvAccess(stream.Context(), req.Table)
	if err != nil {
		return status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(stream.Context(), req.GetTransaction(), &rtOpt)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		logutil.Logger(stream.Context()).Error("fetch transaction error", zap.Error(err))
		return err
	}
	ri, err := queryCtx.GetSession().HandleRead(stream.Context(), req, txn)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		logutil.Logger(stream.Context()).Error("mutate error", zap.Error(err))
		return status.Error(codes.Aborted, err.Error())
	}
	return rs.readStream(stream.Context(), ri, stream.Send)
}
func (rs *RPCServer) readStream(ctx context.Context, ri session.RecordSet, send func(*tspb.PartialResultSet) error) error {
	var (
		row interface{}
		err error
	)
	startTS := time.Now()
	rsm := buildResultSetMetaData(ri.Columns())
	for {
		row, err = ri.Next(ctx)
		if err != nil {
			break
		}
		prs := &tspb.PartialResultSet{
			Metadata: rsm,
			RowCells: row.(*tspb.SliceCell),
		}
		if err = send(prs); err != nil {
			break
		}
		rsm = nil
	}
	durRead := time.Since(startTS)
	if ri.LastErr() != nil || (err != nil && err != io.EOF) {
		streamReadCounterGerneralErr.Inc()
		executeReadDurationGeneralErr.Observe(durRead.Seconds())
		if ri.LastErr() != nil {
			return ri.LastErr()
		}
		return err
	}
	executeReadDurationGeneralOK.Observe(durRead.Seconds())
	streamReadCounterGerneralOK.Inc()
	return nil
}
func (rs *RPCServer) StreamingSparseRead(req *tspb.SparseReadRequest, stream tspb.Tablestore_StreamingSparseReadServer) error {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		return err
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	isRawkv, err := sess.RawkvAccess(stream.Context(), req.Table)
	if err != nil {
		return status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(stream.Context(), req.GetTransaction(), &rtOpt)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		logutil.Logger(stream.Context()).Error("fetch transaction error", zap.Error(err))
		return err
	}
	ctx := sessionctx.SetStreamReadKey(stream.Context())
	ri, err := queryCtx.GetSession().HandleRead(ctx, req, txn)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		logutil.Logger(stream.Context()).Error("stream sparse read error", zap.Error(err))
		return status.Error(codes.Aborted, err.Error())
	}
	return rs.readStream(ctx, ri, stream.Send)
}
func (rs *RPCServer) StreamingSparseScan(req *tspb.SparseScanRequest, stream tspb.Tablestore_StreamingSparseScanServer) error {
	queryCtx, err := rs.getQueryCtxBySessionID(req.Session)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		return err
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	isRawkv, err := sess.RawkvAccess(stream.Context(), req.Table)
	if err != nil {
		return status.Error(codes.Aborted, err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(stream.Context(), req.GetTransaction(), &rtOpt)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		logutil.Logger(stream.Context()).Error("fetch transaction error", zap.Error(err))
		return err
	}
	ctx := sessionctx.SetStreamReadKey(stream.Context())
	ri, err := queryCtx.GetSession().HandleRead(ctx, req, txn)
	if err != nil {
		streamReadCounterGerneralErr.Inc()
		logutil.Logger(stream.Context()).Error("stream sparse read error", zap.Error(err))
		return status.Error(codes.Aborted, err.Error())
	}
	return rs.readStream(ctx, ri, stream.Send)
}
func (rs *RPCServer) BeginTransaction(ctx context.Context, req *tspb.BeginTransactionRequest) (*tspb.Transaction, error) {
	queryCtx, err := rs.getQueryCtxBySessionID(req.GetSession())
	if err != nil {
		return nil, err
	}
	sess := queryCtx.GetSession()
	sess.SetLastActive(time.Now())
	txnSel := &tspb.TransactionSelector{
		Selector: &tspb.TransactionSelector_Begin{
			Begin: req.GetOptions(),
		},
	}
	txn, err := sess.RetrieveTxn(ctx, txnSel, nil)
	if err != nil {
		logutil.Logger(ctx).Error("session retrieve transaction error", zap.Error(err), zap.String("session", req.Session))
		return nil, status.Errorf(codes.Internal, "begin transaction error %v", err)
	}
	transaction := &tspb.Transaction{
		Id: []byte(txn.ID),
	}
	return transaction, nil
}