func (h *HBaseHandler) Increment(ctx context.Context, increment *hbase.TIncrement) (err error) {
	return thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "not implement yet")
}
Increment a cell by the ammount.
Increments can be applied async if hbase.regionserver.thrift.coalesceIncrement is set to true.
False is the default.  Turn to true if you need the extra performance and can accept some
data loss if a thrift server dies with increments still in the queue.

Parameters:
- Increment: The single increment to apply
package hthrift
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"log"
	"math"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"github.com/gogo/protobuf/proto"
	"github.com/pingcap/tidb/util/logutil"
	"github.com/pingcap/tidb/util/mock"
	"github.com/spf13/cast"
	"github.com/uber/jaeger-client-go/thrift"
	tspb "github.com/zhihu/zetta-proto/pkg/tablestore"
	"github.com/zhihu/zetta/pkg/meta"
	"github.com/zhihu/zetta/pkg/model"
	"github.com/zhihu/zetta/tablestore/domain"
	"github.com/zhihu/zetta/tablestore/server/hthrift/hbase"
	"github.com/zhihu/zetta/tablestore/session"
	tst "github.com/zhihu/zetta/tablestore/table"
	"github.com/zhihu/zetta/tablestore/table/tables"
	"go.uber.org/zap"
)
const (
	LATEST_TIMESTAMP = math.MaxInt64
)
func init() {
	rand.Seed(time.Now().UnixNano())
}
type HBaseHandler struct {
	ts            *TServer
	nextScannerID int64
	scannerMap    map[int64]*TScanner
	mu            sync.RWMutex
}
func NewHBaseHandler(ts *TServer) *HBaseHandler {
	return &HBaseHandler{
		ts:         ts,
		scannerMap: make(map[int64]*TScanner),
	}
}
func (h *HBaseHandler) addScanner(scanner *TScanner) int64 {
	scannerID := atomic.AddInt64(&h.nextScannerID, 1)
	h.mu.Lock()
	defer h.mu.Unlock()
	h.scannerMap[scannerID] = scanner
	scanner.scannerID = scannerID
	return scannerID
}
func (h *HBaseHandler) getScanner(id int64) *TScanner {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.scannerMap[int64(id)]
}
func (h *HBaseHandler) removeScanner(id int64) {
	h.mu.Lock()
	defer h.mu.Unlock()
	delete(h.scannerMap, int64(id))
}
func (h *HBaseHandler) EnableTable(ctx context.Context, tableName hbase.Bytes) (err error) {
	return nil
}
func (h *HBaseHandler) DisableTable(ctx context.Context, tableName hbase.Bytes) (err error) {
	return nil
}
func (h *HBaseHandler) IsTableEnabled(ctx context.Context, tableName hbase.Bytes) (r bool, err error) {
	return true, nil
}
func (h *HBaseHandler) Compact(ctx context.Context, tableNameOrRegionName hbase.Bytes) (err error) {
	return nil
}
func (h *HBaseHandler) MajorCompact(ctx context.Context, tableNameOrRegionName hbase.Bytes) (err error) {
	return nil
}
func (h *HBaseHandler) GetTableNames(ctx context.Context) (r []hbase.Text, err error) {
	domain := domain.GetOnlyDomain()
	dbs := domain.InfoSchema().ListDatabases()
	r = []hbase.Text{}
	for _, db := range dbs {
		tables := domain.InfoSchema().ListTablesByDatabase(db.Database)
		for _, t := range tables {
			tableName := db.Database + ":" + t.TableName
			r = append(r, hbase.Text(tableName))
		}
	}
	return r, nil
}
func (h *HBaseHandler) GetColumnDescriptors(ctx context.Context, tableName hbase.Text) (r map[string]*hbase.ColumnDescriptor, err error) {
	db, tn, err := getDBTable(tableName)
	if err != nil {
		return nil, buildIOError(err)
	}
	domain := domain.GetOnlyDomain()
	tt, err := domain.InfoSchema().GetTableByName(db, tn)
	if err != nil {
		logutil.Logger(ctx).Error("prepare zetta table info error", zap.Error(err))
		return nil, buildIOError(err.Error())
	}
	tableMeta := tt.Meta()
	r = make(map[string]*hbase.ColumnDescriptor)
	for _, cf := range tableMeta.ColumnFamilies {
		if !tables.IsFlexible(cf) {
			continue
		}
		colDesc := &hbase.ColumnDescriptor{
			Name:       hbase.Text(cf.Name),
			TimeToLive: 604800,
		}
		colDesc.BlockCacheEnabled = cast.ToBool(cf.Attributes["BLOCKCACHE"])
		colDesc.TimeToLive = cast.ToInt32(cf.Attributes["TTL"])
		colDesc.BloomFilterType = cf.Attributes["BLOOMFILTER"]
		colDesc.Compression = cf.Attributes["COMPRESSION"]
		colDesc.MaxVersions = cast.ToInt32(cf.Attributes["VERSION"])
		colDesc.InMemory = cast.ToBool(cf.Attributes["IN_MEMORY"])
		r[cf.Name] = colDesc
	}
	return r, nil
}
func (h *HBaseHandler) GetTableRegions(ctx context.Context, tableName hbase.Text) (r []*hbase.TRegionInfo, err error) {
	return nil, nil
}
func (h *HBaseHandler) CreateTable(ctx context.Context, tableName hbase.Text, columnFamilies []*hbase.ColumnDescriptor) (err error) {
	db, tn, err := getDBTable(tableName)
	if err != nil {
		return buildIOError(err)
	}
	do := domain.GetOnlyDomain()
	_, err = do.InfoSchema().GetTableMetaByName(db, tn)
	if err == meta.ErrDBNotExists {
		return &hbase.IllegalArgument{
			Message: fmt.Sprintf("%s not exists", db),
		}
	} else if err == nil {
		return &hbase.AlreadyExists{
			Message: fmt.Sprintf("%s already exists", string(tableName)),
		}
	}
	req := &tspb.CreateTableRequest{
		Database: db,
		TableMeta: &tspb.TableMeta{
			Database:       db,
			TableName:      tn,
			ColumnFamilies: []*tspb.ColumnFamilyMeta{},
			Columns: []*tspb.ColumnMeta{
				{Name: "rowkey", ColumnType: &tspb.Type{Code: tspb.TypeCode_BYTES}, Family: "default"},
			},
			PrimaryKey: []string{"rowkey"},
		},
	}
	for _, cf := range columnFamilies {
		if bytes.HasSuffix(cf.Name, []byte(":")) {
			cf.Name = bytes.TrimRight(cf.Name, ":")
		}
		cfMeta := &tspb.ColumnFamilyMeta{
			Name: string(cf.Name),
			Attributes: map[string]string{
				"mode":   "flexible",
				"layout": "sparse",
			},
		}
		cfMeta.Attributes["BLOCKCACHE"] = cast.ToString(cf.BlockCacheEnabled)
		cfMeta.Attributes["BLOOMFILTER"] = cf.BloomFilterType
		cfMeta.Attributes["COMPRESSION"] = cf.Compression
		cfMeta.Attributes["VERSION"] = cast.ToString(cf.MaxVersions)
		cfMeta.Attributes["TTL"] = cast.ToString(cf.TimeToLive)
		cfMeta.Attributes["IN_MEMORY"] = cast.ToString(cf.InMemory)
		if cf.TimeToLive == -1 {
			cfMeta.Attributes["TTL"] = cast.ToString(math.MaxInt32)
		}
		req.TableMeta.ColumnFamilies = append(req.TableMeta.ColumnFamilies, cfMeta)
	}
	tabMeta := model.NewTableMetaFromPbReq(req)
	if err := do.DDL().CreateTable(mock.NewContext(), tabMeta, false); err != nil {
		return &hbase.IOError{
			Message: err.Error(),
		}
	}
	return nil
}
func (h *HBaseHandler) DeleteTable(ctx context.Context, tableName hbase.Text) (err error) {
	return nil
}
func (h *HBaseHandler) Get(ctx context.Context, tableName hbase.Text, row hbase.Text, column hbase.Text, attributes map[string]hbase.Text) (r []*hbase.TCell, err error) {
	return h.GetVerTs(ctx, tableName, row, column, LATEST_TIMESTAMP, 1, attributes)
}
func (h *HBaseHandler) GetVer(ctx context.Context, tableName hbase.Text, row hbase.Text, column hbase.Text, numVersions int32, attributes map[string]hbase.Text) (r []*hbase.TCell, err error) {
	return h.GetVerTs(ctx, tableName, row, column, LATEST_TIMESTAMP, 1, attributes)
}
func (h *HBaseHandler) GetVerTs(ctx context.Context, tableName hbase.Text, row hbase.Text, column hbase.Text, timestamp int64, numVersions int32, attributes map[string]hbase.Text) (r []*hbase.TCell, err error) {
	if column == nil || len(column) == 0 {
		return nil, buildIOError("column should be specific")
	}
	family, col, err := getFamilyQualifier([]byte(column))
	if err != nil {
		return nil, buildIOError(err)
	}
	db, tn, err := getDBTable(tableName)
	if err != nil {
		return nil, buildIOError(err)
	}
	resource, err := h.ts.sessPool.Get()
	if err != nil {
		return nil, buildIOError(err)
	}
	defer h.ts.sessPool.Put(resource)
	txnSel := &tspb.TransactionSelector{
		Selector: &tspb.TransactionSelector_SingleUse{
			SingleUse: &tspb.TransactionOptions{
				Mode: &tspb.TransactionOptions_ReadWrite_{},
			},
		},
	}
	sess := resource.(session.Session)
	sess.SetDB(db)
	isRawkv, err := sess.RawkvAccess(ctx, tn)
	if err != nil {
		return nil, buildIOError(err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, txnSel, &rtOpt)
	if err != nil {
		return nil, buildIOError(err)
	}
	rkey := &tspb.ListValue{
		Values: []*tspb.Value{
			{Kind: &tspb.Value_BytesValue{BytesValue: []byte(row)}},
		},
	}
	tsrow := &tspb.Row{
		Keys:       rkey,
		Qualifiers: nil,
	}
	if col != "" {
		tsrow.Qualifiers = []string{col}
	} else {
		tsrow.Qualifiers = []string{}
	}
	req := &tspb.SparseReadRequest{
		Table:  tn,
		Family: family,
		Rows:   []*tspb.Row{tsrow},
	}
	ri, err := sess.HandleRead(ctx, req, txn)
	if err != nil {
		logutil.BgLogger().Error("getVerTs error", zap.Error(err))
		return nil, thrift.NewTApplicationException(thrift.INTERNAL_ERROR, err.Error())
	}
	r = []*hbase.TCell{}
	for {
		rsliceCell, err := ri.Next(ctx)
		if err != nil {
			break
		}
		rscell := rsliceCell.(*tspb.SliceCell)
		for _, cell := range rscell.Cells {
			tcell := &hbase.TCell{Value: cell.Value.GetBytesValue()}
			r = append(r, tcell)
		}
	}
	if ri.LastErr() != nil {
		return nil, buildIOError(ri.LastErr())
	}
	return r, nil
}
func (h *HBaseHandler) GetRow(ctx context.Context, tableName hbase.Text, row hbase.Text, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	return h.GetRowWithColumnsTs(ctx, tableName, row, nil, LATEST_TIMESTAMP, attributes)
}
func (h *HBaseHandler) GetRowWithColumns(ctx context.Context, tableName hbase.Text, row hbase.Text, columns []hbase.Text, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	return h.GetRowWithColumnsTs(ctx, tableName, row, columns, LATEST_TIMESTAMP, attributes)
}
func (h *HBaseHandler) GetRowTs(ctx context.Context, tableName hbase.Text, row hbase.Text, timestamp int64, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	return h.GetRowWithColumnsTs(ctx, tableName, row, nil, timestamp, attributes)
}
func (h *HBaseHandler) GetRowWithColumnsTs(ctx context.Context, tableName hbase.Text, row hbase.Text, columns []hbase.Text, timestamp int64, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	db, tn, err := getDBTable(tableName)
	if err != nil {
		return nil, buildIOError(err)
	}
	reqs, err := buildSparseReadReq(tn, []hbase.Text{row}, columns)
	if err != nil {
		return nil, buildIOError(err)
	}
	resource, err := h.ts.sessPool.Get()
	if err != nil {
		return nil, buildIOError(err)
	}
	defer h.ts.sessPool.Put(resource)
	sess := resource.(session.Session)
	sess.SetDB(db)
	txnSel := &tspb.TransactionSelector{
		Selector: &tspb.TransactionSelector_SingleUse{
			SingleUse: &tspb.TransactionOptions{
				Mode: &tspb.TransactionOptions_ReadWrite_{},
			},
		},
	}
	isRawkv, err := sess.RawkvAccess(ctx, tn)
	if err != nil {
		return nil, buildIOError(err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, txnSel, &rtOpt)
	if err != nil {
		return nil, buildIOError(err)
	}
	results := make([]*hbase.TRowResult_, 1)
	result := &hbase.TRowResult_{
		Row:     row,
		Columns: make(map[string]*hbase.TCell),
	}
	for _, req := range reqs {
		ri, err := sess.HandleRead(ctx, req, txn)
		if err != nil {
			logutil.BgLogger().Error("GetRowWithColumnsTs", zap.Error(err))
			return nil, buildIOError(err)
		}
		for {
			rsliceCell, err := ri.Next(ctx)
			if err != nil {
				break
			}
			rscell := rsliceCell.(*tspb.SliceCell)
			for _, cell := range rscell.Cells {
				tcell := &hbase.TCell{Value: cell.Value.GetBytesValue()}
				columnName := cell.Family + ":" + cell.Column
				result.Columns[columnName] = tcell
			}
		}
		if ri.LastErr() != nil {
			return nil, buildIOError(ri.LastErr())
		}
	}
	results[0] = result
	return results, nil
}
func (h *HBaseHandler) GetRows(ctx context.Context, tableName hbase.Text, rows []hbase.Text, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	return h.GetRowsWithColumnsTs(ctx, tableName, rows, nil, LATEST_TIMESTAMP, attributes)
}
func (h *HBaseHandler) GetRowsWithColumns(ctx context.Context, tableName hbase.Text, rows []hbase.Text, columns []hbase.Text, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	return h.GetRowsWithColumnsTs(ctx, tableName, rows, columns, LATEST_TIMESTAMP, attributes)
}
func (h *HBaseHandler) GetRowsTs(ctx context.Context, tableName hbase.Text, rows []hbase.Text, timestamp int64, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	return h.GetRowsWithColumnsTs(ctx, tableName, rows, nil, LATEST_TIMESTAMP, attributes)
}
func (h *HBaseHandler) GetRowsWithColumnsTs(ctx context.Context, tableName hbase.Text, rows []hbase.Text, columns []hbase.Text, timestamp int64, attributes map[string]hbase.Text) (r []*hbase.TRowResult_, err error) {
	db, tn, err := getDBTable(tableName)
	if err != nil {
		return nil, buildIOError(err)
	}
	reqs, err := buildSparseReadReq(tn, rows, columns)
	if err != nil {
		return nil, buildIOError(err)
	}
	resource, err := h.ts.sessPool.Get()
	if err != nil {
		return nil, buildIOError(err)
	}
	defer h.ts.sessPool.Put(resource)
	sess := resource.(session.Session)
	sess.SetDB(db)
	txnSel := &tspb.TransactionSelector{
		Selector: &tspb.TransactionSelector_SingleUse{
			SingleUse: &tspb.TransactionOptions{
				Mode: &tspb.TransactionOptions_ReadWrite_{},
			},
		},
	}
	isRawkv, err := sess.RawkvAccess(ctx, tn)
	if err != nil {
		return nil, buildIOError(err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: false,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, txnSel, &rtOpt)
	if err != nil {
		return nil, buildIOError(err)
	}
	results := make([]*hbase.TRowResult_, 0)
	sliceCells := make([]*tspb.SliceCell, 0)
	rowCells := make(map[string]map[string]*hbase.TCell)
	for _, req := range reqs {
		ri, err := sess.HandleRead(ctx, req, txn)
		if err != nil {
			logutil.BgLogger().Error("GetRowsWithColumnsTs", zap.Error(err))
			return nil, buildIOError(err)
		}
		for {
			rsliceCell, err := ri.Next(ctx)
			if err != nil {
				break
			}
			rscell := rsliceCell.(*tspb.SliceCell)
			sliceCells = append(sliceCells, rscell)
		}
		if ri.LastErr() != nil {
			return nil, buildIOError(ri.LastErr())
		}
	}
	for _, scell := range sliceCells {
		rowkey := scell.PrimaryKeys[0].GetBytesValue()
		columns, ok := rowCells[string(rowkey)]
		if !ok {
			columns = map[string]*hbase.TCell{}
			rowCells[string(rowkey)] = columns
		}
		for _, cell := range scell.Cells {
			tcell := &hbase.TCell{Value: cell.Value.GetBytesValue()}
			columnName := cell.Family + ":" + cell.Column
			columns[columnName] = tcell
		}
	}
	for _, row := range rows {
		columns := rowCells[string(row)]
		if columns == nil {
			continue
		}
		result := &hbase.TRowResult_{
			Row:     row,
			Columns: columns,
		}
		results = append(results, result)
	}
	return results, nil
}
func (h *HBaseHandler) MutateRow(ctx context.Context, tableName hbase.Text, row hbase.Text, mutations []*hbase.Mutation, attributes map[string]hbase.Text) (err error) {
	log.Printf("MutateRow Table: %v, Row: %v \n", string(tableName), string(row))
	return h.MutateRowTs(ctx, tableName, row, mutations, LATEST_TIMESTAMP, attributes)
}
func (h *HBaseHandler) MutateRowTs(ctx context.Context, tableName hbase.Text, row hbase.Text, mutations []*hbase.Mutation, timestamp int64, attributes map[string]hbase.Text) (err error) {
	rowBatchs := []*hbase.BatchMutation{
		{
			Row:       row,
			Mutations: mutations,
		},
	}
	return h.MutateRowsTs(ctx, tableName, rowBatchs, timestamp, attributes)
}
func (h *HBaseHandler) MutateRows(ctx context.Context, tableName hbase.Text, rowBatches []*hbase.BatchMutation, attributes map[string]hbase.Text) (err error) {
	return h.MutateRowsTs(ctx, tableName, rowBatches, LATEST_TIMESTAMP, attributes)
}
func (h *HBaseHandler) MutateRowsTs(ctx context.Context, tableName hbase.Text, rowBatches []*hbase.BatchMutation, timestamp int64, attributes map[string]hbase.Text) (err error) {
	db, table, err := getDBTable([]byte(tableName))
	if err != nil {
		return buildIOError(err)
	}
	mutReq := &tspb.MutationRequest{
		Table:     table,
		Mutations: make([]*tspb.Mutation, 0),
	}
	if rowBatches != nil {
		for _, batch := range rowBatches {
			for _, mutation := range batch.GetMutations() {
				zmut, err := buildZettaMutation(table, batch.GetRow(), mutation)
				if err != nil {
					log.Println(err)
					return err
				}
				mutReq.Mutations = append(mutReq.Mutations, zmut)
			}
		}
	}
	resource, err := h.ts.sessPool.Get()
	if err != nil {
		return buildIOError(err)
	}
	defer h.ts.sessPool.Put(resource)
	txnSel := &tspb.TransactionSelector{
		Selector: &tspb.TransactionSelector_SingleUse{
			SingleUse: &tspb.TransactionOptions{
				Mode: &tspb.TransactionOptions_ReadWrite_{},
			},
		},
	}
	sess := resource.(session.Session)
	sess.SetDB(db)
	isRawkv, err := sess.RawkvAccess(ctx, table)
	if err != nil {
		return buildIOError(err.Error())
	}
	rtOpt := session.RetrieveTxnOpt{
		Committable: true,
		IsRawKV:     isRawkv,
	}
	txn, err := sess.RetrieveTxn(ctx, txnSel, &rtOpt)
	if err != nil {
		return buildIOError(err)
	}
	startTs := time.Now()
	_, err = sess.HandleMutate(ctx, mutReq, txn)
	if err != nil {
		return buildIOError(err)
	}
	duration := time.Now().Sub(startTs)
	if rand.Intn(1000)%10 == 0 {
		logutil.BgLogger().Info("handle mutate consume time", zap.Duration("dur", duration))
	}
	return nil
}
func (h *HBaseHandler) AtomicIncrement(ctx context.Context, tableName hbase.Text, row hbase.Text, column hbase.Text, value int64) (r int64, err error) {
	return -1, thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "not implement yet")
}
func (h *HBaseHandler) DeleteAll(ctx context.Context, tableName hbase.Text, row hbase.Text, column hbase.Text, attributes map[string]hbase.Text) (err error) {
	return thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "not implement yet")
}
func (h *HBaseHandler) DeleteAllTs(ctx context.Context, tableName hbase.Text, row hbase.Text, column hbase.Text, timestamp int64, attributes map[string]hbase.Text) (err error) {
	return thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "not implement yet")
}
func (h *HBaseHandler) DeleteAllRow(ctx context.Context, tableName hbase.Text, row hbase.Text, attributes map[string]hbase.Text) (err error) {
	return thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "not implement yet")
}