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")
}
Delete all cells that match the passed row and column and whose
timestamp is equal-to or older than the passed timestamp.

Parameters:
- TableName: name of table
- Row: Row to update
- Column: name of column whose value is to be deleted
- Timestamp: timestamp
- Attributes: Delete attributes
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")
}