api.go

package tablestore

import (
	"bytes"
	"crypto/md5"
	"encoding/base64"
	"errors"
	"fmt"
	"hash/crc32"
	"io"
	"io/ioutil"
	"math/rand"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/aliyun/aliyun-tablestore-go-sdk/common"
	"github.com/aliyun/aliyun-tablestore-go-sdk/tablestore/otsprotocol"
	Fieldvalues "github.com/aliyun/aliyun-tablestore-go-sdk/tablestore/timeseries/flatbuffer"
	"github.com/golang/protobuf/proto"
	lruCache "github.com/hashicorp/golang-lru"
)

const (
	userAgent = "aliyun-tablestore-sdk-golang/4.0.2"

	createTableUri                     = "/CreateTable"
	listTableUri                       = "/ListTable"
	deleteTableUri                     = "/DeleteTable"
	describeTableUri                   = "/DescribeTable"
	updateTableUri                     = "/UpdateTable"
	putRowUri                          = "/PutRow"
	deleteRowUri                       = "/DeleteRow"
	getRowUri                          = "/GetRow"
	updateRowUri                       = "/UpdateRow"
	batchGetRowUri                     = "/BatchGetRow"
	batchWriteRowUri                   = "/BatchWriteRow"
	getRangeUri                        = "/GetRange"
	listStreamUri                      = "/ListStream"
	describeStreamUri                  = "/DescribeStream"
	getShardIteratorUri                = "/GetShardIterator"
	getStreamRecordUri                 = "/GetStreamRecord"
	computeSplitPointsBySizeRequestUri = "/ComputeSplitPointsBySize"
	searchUri                          = "/Search"
	createSearchIndexUri               = "/CreateSearchIndex"
	updateSearchIndexUri               = "/UpdateSearchIndex"
	listSearchIndexUri                 = "/ListSearchIndex"
	deleteSearchIndexUri               = "/DeleteSearchIndex"
	describeSearchIndexUri             = "/DescribeSearchIndex"
	computeSplitsUri                   = "/ComputeSplits"
	parallelScanUri                    = "/ParallelScan"
	sqlQueryUri                        = "/SQLQuery"

	createIndexUri = "/CreateIndex"
	dropIndexUri   = "/DropIndex"

	createDeliveryTaskUri   = "/CreateDeliveryTask"
	deleteDeliveryTaskUri   = "/DeleteDeliveryTask"
	updateDeliveryTaskUri   = "/UpdateDeliveryTask"
	describeDeliveryTaskUri = "/DescribeDeliveryTask"
	listDeliveryTaskUri     = "/ListDeliveryTask"

	createlocaltransactionuri = "/StartLocalTransaction"
	committransactionuri      = "/CommitTransaction"
	aborttransactionuri       = "/AbortTransaction"

	adddefinedcolumnuri    = "/AddDefinedColumn"
	deletedefinedcolumnuri = "/DeleteDefinedColumn"

	createTimeseriesTable             = "/CreateTimeseriesTable"
	putTimeseriesData                 = "/PutTimeseriesData"
	getTimeseriesData                 = "/GetTimeseriesData"
	queryTimeseriesMeta               = "/QueryTimeseriesMeta"
	listTimeseriesTable               = "/ListTimeseriesTable"
	deleteTimeseriesTable             = "/DeleteTimeseriesTable"
	describeTimeseriesTable           = "/DescribeTimeseriesTable"
	updateTimeseriesTable             = "/UpdateTimeseriesTable"
	updateTimeseriesMeta              = "/UpdateTimeseriesMeta"
	deleteTimeseriesMeta              = "/DeleteTimeseriesMeta"
	createTimeseriesAnalyticalStore   = "/CreateTimeseriesAnalyticalStore"
	deleteTimeseriesAnalyticalStore   = "/DeleteTimeseriesAnalyticalStore"
	describeTimeseriesAnalyticalStore = "/DescribeTimeseriesAnalyticalStore"
	updateTimeseriesAnalyticalStore   = "/UpdateTimeseriesAnalyticalStore"
)

// RowsSerializeType is used for tests only.
var RowsSerializeType = otsprotocol.RowsSerializeType_RST_FLAT_BUFFER

// Constructor: to create the client of TableStore service.
// 构造函数：创建表格存储服务的客户端。
//
// @param endPoint The address of TableStore service. 表格存储服务地址。
// @param instanceName
// @param accessId The Access ID. 用于标示用户的ID。
// @param accessKey The Access Key. 用于签名和验证的密钥。
// @param options set client config
func NewClient(endPoint, instanceName, accessKeyId, accessKeySecret string, options ...ClientOption) *TableStoreClient {
	client := NewClientWithConfig(endPoint, instanceName, accessKeyId, accessKeySecret, "", nil, options...)
	// client options parse
	for _, option := range options {
		option(client)
	}

	return client
}

type GetHttpClient func() IHttpClient

var currentGetHttpClientFunc GetHttpClient = func() IHttpClient {
	return &TableStoreHttpClient{}
}

// Constructor: to create the client of OTS service. 传入config
// 构造函数：创建OTS服务的客户端。
func NewClientWithConfig(endPoint, instanceName, accessKeyId, accessKeySecret string, securityToken string, config *TableStoreConfig, options ...ClientOption) *TableStoreClient {
	tableStoreClient := new(TableStoreClient)
	tableStoreClient.internalClient = new(internalClient)
	tableStoreClient.endPoint = endPoint
	tableStoreClient.instanceName = instanceName
	tableStoreClient.accessKeyId = accessKeyId
	tableStoreClient.accessKeySecret = accessKeySecret
	tableStoreClient.securityToken = securityToken
	tableStoreClient.KeepDefaultRetryStrategyWhileUsingCustomizedRetryFunc = true
	provider := &common.DefaultCredentialsProvider{AccessKeyID: accessKeyId, AccessKeySecret: accessKeySecret, SecurityToken: securityToken}
	tableStoreClient.credentialsProvider = provider
	for _, option := range options {
		option(tableStoreClient)
	}

	if config == nil {
		config = NewDefaultTableStoreConfig()
	}
	tableStoreClient.config = config
	var tableStoreTransportProxy http.RoundTripper
	if config.Transport != nil {
		tableStoreTransportProxy = config.Transport
	} else {
		tableStoreTransportProxy = &http.Transport{
			MaxIdleConnsPerHost: config.MaxIdleConnections,
			IdleConnTimeout:     config.IdleConnTimeout,
			Dial: (&net.Dialer{
				Timeout: config.HTTPTimeout.ConnectionTimeout,
			}).Dial,
		}
	}

	tableStoreClient.httpClient = currentGetHttpClientFunc()

	httpClient := &http.Client{
		Transport: tableStoreTransportProxy,
		Timeout:   tableStoreClient.config.HTTPTimeout.RequestTimeout,
	}
	tableStoreClient.httpClient.New(httpClient)

	tableStoreClient.mu = &sync.Mutex{}
	tableStoreClient.random = rand.New(rand.NewSource(time.Now().Unix()))

	return tableStoreClient
}

func NewTimeseriesClient(endPoint, instanceName, accessKeyId, accessKeySecret string, options ...TimeseriesClientOption) *TimeseriesClient {
	timeseriesClient := NewTimeseriesClientWithConfig(endPoint, instanceName, accessKeyId, accessKeySecret, "", nil, nil)

	for _, option := range options {
		option(timeseriesClient)
	}

	return timeseriesClient
}

func NewTimeseriesClientWithConfig(endPoint, instanceName, accessKeyId, accessKeySecret string, securityToken string, config *TableStoreConfig, timeseriesConfiguration *TimeseriesConfiguration, options ...TimeseriesClientOption) *TimeseriesClient {
	timeseriesClient := new(TimeseriesClient)
	timeseriesClient.internalClient = new(internalClient)
	timeseriesClient.endPoint = endPoint
	timeseriesClient.instanceName = instanceName
	timeseriesClient.accessKeyId = accessKeyId
	timeseriesClient.accessKeySecret = accessKeySecret
	timeseriesClient.securityToken = securityToken
	timeseriesClient.KeepDefaultRetryStrategyWhileUsingCustomizedRetryFunc = true
	provider := &common.DefaultCredentialsProvider{AccessKeyID: accessKeyId, AccessKeySecret: accessKeySecret, SecurityToken: securityToken}
	timeseriesClient.credentialsProvider = provider
	for _, option := range options {
		option(timeseriesClient)
	}

	if config == nil {
		config = NewDefaultTableStoreConfig()
	}
	timeseriesClient.config = config
	if timeseriesConfiguration == nil {
		timeseriesConfiguration = NewTimeseriesConfiguration()
	}
	timeseriesClient.timeseriesConfiguration = timeseriesConfiguration
	var tableStoreTransportProxy http.RoundTripper
	if config.Transport != nil {
		tableStoreTransportProxy = config.Transport
	} else {
		tableStoreTransportProxy = &http.Transport{
			MaxIdleConnsPerHost: config.MaxIdleConnections,
			IdleConnTimeout:     config.IdleConnTimeout,
			Dial: (&net.Dialer{
				Timeout: config.HTTPTimeout.ConnectionTimeout,
			}).Dial,
		}
	}

	timeseriesClient.httpClient = currentGetHttpClientFunc()

	httpClient := &http.Client{
		Transport: tableStoreTransportProxy,
		Timeout:   timeseriesClient.config.HTTPTimeout.RequestTimeout,
	}
	timeseriesClient.httpClient.New(httpClient)

	timeseriesClient.mu = &sync.Mutex{}
	timeseriesClient.random = rand.New(rand.NewSource(time.Now().Unix()))

	timeseriesMetaCache, _ := lruCache.New(timeseriesClient.timeseriesConfiguration.metaCacheMaxDataSize)
	timeseriesClient.SetTimeseriesMetaCache(timeseriesMetaCache)
	return timeseriesClient
}

func NewClientWithExternalHeader(endPoint, instanceName, accessKeyId, accessKeySecret string, securityToken string, config *TableStoreConfig, header map[string]string) *TableStoreClient {
	tableStoreClient := NewClientWithConfig(endPoint, instanceName, accessKeyId, accessKeySecret, securityToken, config)
	tableStoreClient.externalHeader = header
	return tableStoreClient
}

type RetryNotify func(traceId, requestId string, err error, action string, backoffDuration time.Duration)

// 请求服务端
func (internalClient *internalClient) doRequestWithRetry(uri string, req, resp proto.Message, responseInfo *ResponseInfo, extraInfo ExtraRequestInfo) error {
	end := time.Now().Add(internalClient.config.MaxRetryTime)
	url := fmt.Sprintf("%s%s", internalClient.endPoint, uri)
	/* request body */
	var body []byte
	var err error
	if req != nil {
		body, err = proto.Marshal(req)
		if err != nil {
			return err
		}
	} else {
		body = nil
	}

	var value int64
	var i uint
	var respBody []byte
	var requestId string
	for i = 0; ; i++ {
		respBody, err, requestId = internalClient.doRequest(url, uri, body, resp, extraInfo)
		responseInfo.RequestId = requestId

		if err == nil {
			break
		} else {
			value = internalClient.getNextPause(err, i, end, value, uri)

			// fmt.Println("hit retry", uri, err, *e.Code, Value)
			if value <= 0 {
				return err
			}

			if internalClient.RetryNotify != nil {
				traceId := ""
				if extraInfo.userTraceID != nil {
					traceId = *extraInfo.userTraceID
				}
				internalClient.RetryNotify(traceId, requestId, err, uri, time.Duration(value)*time.Millisecond)
			}

			time.Sleep(time.Duration(value) * time.Millisecond)
		}
	}

	if len(respBody) == 0 {
		return nil
	}

	err = proto.Unmarshal(respBody, resp)
	if err != nil {
		return fmt.Errorf("decode resp failed: %s", err)
	}

	return nil
}

func (internalClient *internalClient) doBatchRequestWithRetry(uri string, req, resp proto.Message, responseInfo *ResponseInfo, extraInfo ExtraRequestInfo) error {
	end := time.Now().Add(internalClient.config.MaxRetryTime)
	url := fmt.Sprintf("%s%s", internalClient.endPoint, uri)
	/* request body */
	body, err := proto.Marshal(req)
	if req != nil && err != nil {
		return err
	}

	var value int64
	var respBody []byte
	var requestId string
	for i := uint(0); ; i++ {

		respBody, err, requestId = internalClient.doRequest(url, uri, body, resp, extraInfo)
		responseInfo.RequestId = requestId

		if err != nil {
			value = internalClient.getNextPause(err, i, end, value, uri)

			// fmt.Println("hit retry", uri, err, *e.Code, Value)
			if value <= 0 {
				return err
			}

			if internalClient.RetryNotify != nil {
				traceId := ""
				if extraInfo.userTraceID != nil {
					traceId = *extraInfo.userTraceID
				}
				internalClient.RetryNotify(traceId, requestId, err, uri, time.Duration(value)*time.Millisecond)
			}

			time.Sleep(time.Duration(value) * time.Millisecond)
		} else {
			if len(respBody) == 0 {
				return nil
			}
			if err = mergeBatchResponse(resp, respBody, uri); err != nil {
				return err
			}
			var partitionFailedCouldRetry bool
			body, partitionFailedCouldRetry, err = generateBatchRequest(resp, req, uri)
			if err != nil || !partitionFailedCouldRetry {
				return err
			}

			value = internalClient.getPartitionFailedNextPause(i, end, value)
			if value <= 0 {
				return nil
			}

			traceId := ""
			if extraInfo.userTraceID != nil {
				traceId = *extraInfo.userTraceID
			}
			if internalClient.RetryNotify != nil {
				internalClient.RetryNotify(traceId, requestId, err, uri, time.Duration(value)*time.Millisecond)
			}

			time.Sleep(time.Duration(value) * time.Millisecond)
		}
	}
}

func mergeBatchResponse(previousResp proto.Message, currentRespBody []byte, uri string) error {
	// merge result of currentRespBody into previousResp
	var err error
	switch uri {
	case batchWriteRowUri:
		// decode currentRespBody
		if previousResp == nil ||
			previousResp.(*otsprotocol.BatchWriteRowResponse).GetTables() == nil { // first response, no need to merge response
			err = proto.Unmarshal(currentRespBody, previousResp)
			return err
		} else { // merge response
			currentResp := new(otsprotocol.BatchWriteRowResponse)
			err = proto.Unmarshal(currentRespBody, currentResp)
			if err != nil {
				return fmt.Errorf("decode resp failed: %s", err)
			}
			// parse currentResp into map
			rowResultMap := map[string][]*otsprotocol.RowInBatchWriteRowResponse{}
			for _, currentTable := range currentResp.Tables {
				rowResultMap[currentTable.GetTableName()] = currentTable.GetRows()
			}
			// update failed partition in previousResponse
			for _, previousTable := range previousResp.(*otsprotocol.BatchWriteRowResponse).Tables {
				index := 0
				for _, previousRowResult := range previousTable.GetRows() {
					if !previousRowResult.GetIsOk() { // previous response failed
						tmpRowResult := rowResultMap[previousTable.GetTableName()][index]
						index++
						if tmpRowResult.GetIsOk() { // new response success
							*previousRowResult.IsOk = true
							previousRowResult.Row = tmpRowResult.GetRow()
							previousRowResult.Consumed = tmpRowResult.GetConsumed()
						} else {
							*previousRowResult.IsOk = false // new response failed
							previousRowResult.Error = tmpRowResult.GetError()
							previousRowResult.Row = tmpRowResult.GetRow()
						}
					}
				}
			}
			return nil
		}
	case batchGetRowUri:
		// decode currentRespBody
		if previousResp == nil ||
			previousResp.(*otsprotocol.BatchGetRowResponse).GetTables() == nil { // first response, no need to merge response
			err = proto.Unmarshal(currentRespBody, previousResp)
			return err
		} else { // merge response
			currentResp := new(otsprotocol.BatchGetRowResponse)
			err = proto.Unmarshal(currentRespBody, currentResp)
			if err != nil {
				return fmt.Errorf("decode resp failed: %s", err)
			}
			rowResultMap := map[string][]*otsprotocol.RowInBatchGetRowResponse{}
			for _, currentTable := range currentResp.Tables {
				rowResultMap[currentTable.GetTableName()] = currentTable.GetRows()
			}
			// update failed partition in previousResponse
			for _, previousTable := range previousResp.(*otsprotocol.BatchGetRowResponse).Tables {
				index := 0
				for _, previousRowResult := range previousTable.GetRows() {
					if !previousRowResult.GetIsOk() { // previous response failed
						tmpRowResult := rowResultMap[previousTable.GetTableName()][index]
						index++
						if tmpRowResult.GetIsOk() { // new response success
							*previousRowResult.IsOk = true
							previousRowResult.Row = tmpRowResult.GetRow()
							previousRowResult.Consumed = tmpRowResult.GetConsumed()
						} else {
							*previousRowResult.IsOk = false // new response failed
							previousRowResult.Error = tmpRowResult.GetError()
							previousRowResult.Row = tmpRowResult.GetRow()
						}
					}
				}
			}
			return nil
		}
	default:
		return fmt.Errorf("unsupport batch handle function: %s", uri)
	}
}

func generateBatchRequest(resp proto.Message, originRequest proto.Message, uri string) ([]byte, bool, error) {
	// if all success or failure could not retry: return nil, false, nil
	// if partition failed and could retry: return nextRequest, true, nil
	respContainsFailedRows := false
	switch uri {
	case batchWriteRowUri:
		tmpReq := new(otsprotocol.BatchWriteRowRequest)
		var tablesInBatch []*otsprotocol.TableInBatchWriteRowRequest
		// parse originRequest into map
		rowRequestMap := map[string][]*otsprotocol.RowInBatchWriteRowRequest{}
		for _, tableRequest := range originRequest.(*otsprotocol.BatchWriteRowRequest).Tables {
			rowRequestMap[tableRequest.GetTableName()] = tableRequest.GetRows()
		}

		for _, table := range resp.(*otsprotocol.BatchWriteRowResponse).Tables {
			tmpTable := new(otsprotocol.TableInBatchWriteRowRequest)
			tmpTable.TableName = proto.String(table.GetTableName())
			index := 0
			for _, rowResult := range table.GetRows() {
				if !rowResult.GetIsOk() {
					// 只有所有失败的行可重试时，批量操作才可以重试
					if !ShouldRetryViaErrorAndAction(rowResult.GetError().GetCode(), rowResult.GetError().GetMessage(), uri) {
						return nil, false, nil
					}
					respContainsFailedRows = true
					tmpTable.Rows = append(tmpTable.Rows, rowRequestMap[table.GetTableName()][index])
				}
				index++
			}
			if len(tmpTable.GetRows()) != 0 {
				tablesInBatch = append(tablesInBatch, tmpTable)
			}
		}
		tmpReq.Tables = tablesInBatch
		tmpReq.IsAtomic = proto.Bool(originRequest.(*otsprotocol.BatchWriteRowRequest).GetIsAtomic())
		// TODO: set TransactionId after go sdk support transaction.
		newRequest, err := proto.Marshal(tmpReq)
		return newRequest, respContainsFailedRows, err
	case batchGetRowUri:
		tmpReq := new(otsprotocol.BatchGetRowRequest)
		// parse originRequest into map
		rowRequestMap := map[string]*otsprotocol.TableInBatchGetRowRequest{}
		for _, rowRequest := range originRequest.(*otsprotocol.BatchGetRowRequest).Tables {
			rowRequestMap[rowRequest.GetTableName()] = rowRequest
		}

		for _, table := range resp.(*otsprotocol.BatchGetRowResponse).Tables {
			tableRequest := new(otsprotocol.TableInBatchGetRowRequest)
			tableRequest.TableName = rowRequestMap[table.GetTableName()].TableName
			tableRequest.Token = rowRequestMap[table.GetTableName()].GetToken()
			tableRequest.ColumnsToGet = rowRequestMap[table.GetTableName()].ColumnsToGet
			tableRequest.TimeRange = rowRequestMap[table.GetTableName()].TimeRange
			tableRequest.MaxVersions = rowRequestMap[table.GetTableName()].MaxVersions
			tableRequest.CacheBlocks = rowRequestMap[table.GetTableName()].CacheBlocks
			tableRequest.ColumnsToGet = rowRequestMap[table.GetTableName()].ColumnsToGet
			tableRequest.Filter = rowRequestMap[table.GetTableName()].Filter
			tableRequest.StartColumn = rowRequestMap[table.GetTableName()].StartColumn
			tableRequest.EndColumn = rowRequestMap[table.GetTableName()].EndColumn

			index := 0
			for _, rowResult := range table.GetRows() {
				if !rowResult.GetIsOk() {
					// 只有所有失败的行可重试时，批量操作才可以重试
					if !ShouldRetryViaErrorAndAction(rowResult.GetError().GetCode(), rowResult.GetError().GetMessage(), uri) {
						return nil, false, nil
					}

					respContainsFailedRows = true
					primaryKey := rowRequestMap[table.GetTableName()].GetPrimaryKey()[index]
					if primaryKey == nil {
						return nil, respContainsFailedRows, fmt.Errorf("can not find table '%s' with index %d", table.GetTableName(), index)
					}
					tableRequest.PrimaryKey = append(tableRequest.PrimaryKey, primaryKey)
				}
				index++
			}
			if len(tableRequest.PrimaryKey) != 0 {
				tmpReq.Tables = append(tmpReq.Tables, tableRequest)
			}
		}
		newRequest, err := proto.Marshal(tmpReq)
		return newRequest, respContainsFailedRows, err
	default:
		return nil, respContainsFailedRows, fmt.Errorf("unsupport generate batch request handle function: %s", uri)
	}
}

func (internalClient *internalClient) getPartitionFailedNextPause(count uint, end time.Time, lastInterval int64) int64 {
	if internalClient.config.RetryTimes <= count || time.Now().After(end) {
		return 0
	}

	newInterval := internalClient.computeNewRetryInterval(lastInterval)
	return newInterval
}

func (internalClient *internalClient) getNextPause(err error, count uint, end time.Time, lastInterval int64, action string) int64 {
	if internalClient.config.RetryTimes <= count || time.Now().After(end) {
		return 0
	}
	var retry bool
	if otsErr, ok := err.(*OtsError); ok {
		retry = internalClient.shouldRetry(otsErr.Code, otsErr.Message, action, otsErr.HttpStatusCode)
	} else {
		if err == io.EOF || err == io.ErrUnexpectedEOF || //retry on special net error contains EOF or reset
			strings.Contains(err.Error(), io.EOF.Error()) ||
			strings.Contains(err.Error(), "server closed idle connection") ||
			strings.Contains(err.Error(), "Connection reset by peer") ||
			strings.Contains(err.Error(), "connection reset by peer") {
			// server already close this connection, no need to delay
			return 1
		} else if strings.Contains(err.Error(), "connection refused") {
			retry = true
		} else if nErr, ok := err.(net.Error); ok {
			retry = nErr.Temporary()
		}
	}
	if retry {
		newInterval := internalClient.computeNewRetryInterval(lastInterval)
		return newInterval
	}
	return 0
}

func (internalClient *internalClient) computeNewRetryInterval(lastInterval int64) int64 {
	defaultRetryInterval := internalClient.config.DefaultRetryInterval / time.Millisecond
	if defaultRetryInterval <= 0 {
		defaultRetryInterval = DefaultRetryInterval
	}
	maxRetryInterval := internalClient.config.MaxRetryInterval / time.Millisecond
	if maxRetryInterval <= 0 {
		maxRetryInterval = MaxRetryInterval
	}
	// lock/unlock when accessing the rand from a goroutine
	internalClient.mu.Lock()
	value := lastInterval*2 + internalClient.random.Int63n(int64(defaultRetryInterval) - 1) + 1
	internalClient.mu.Unlock()
	if value > int64(maxRetryInterval) {
		value = int64(maxRetryInterval)
	}
	return value
}

func (internalClient *internalClient) postReq(req *http.Request, url string) ([]byte, error, string) {
	resp, err := internalClient.httpClient.Do(req)
	if err != nil {
		return nil, err, ""
	}
	defer resp.Body.Close()

	reqId := getRequestId(resp)
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, err, reqId
	}

	if (resp.StatusCode >= 200 && resp.StatusCode < 300) == false {
		var retErr *OtsError
		perr := new(otsprotocol.Error)
		errUm := proto.Unmarshal(body, perr)
		if errUm != nil {
			retErr = rawHttpToOtsError(resp.StatusCode, body, reqId)
		} else {
			retErr = pbErrToOtsError(resp.StatusCode, perr, reqId)
		}
		return nil, retErr, reqId
	}

	return body, nil, reqId
}

func (internalClient *internalClient) shouldRetry(errorCode string, errorMsg string, action string, statusCode int) bool {
	if internalClient.CustomizedRetryFunc != nil {
		if internalClient.CustomizedRetryFunc(errorCode, errorMsg, action, statusCode) == true {
			return true
		} else if !internalClient.KeepDefaultRetryStrategyWhileUsingCustomizedRetryFunc {
			return false
		}
	}
	return ShouldRetryViaErrorAndAction(errorCode, errorMsg, action)
}

func ShouldRetryViaErrorAndAction(errorCode string, errorMsg string, action string) bool {
	if retryNotMatterActions(errorCode, errorMsg) == true {
		return true
	}

	if isIdempotent(action) &&
		(errorCode == STORAGE_TIMEOUT || errorCode == INTERNAL_SERVER_ERROR || errorCode == SERVER_UNAVAILABLE) {
		return true
	}
	return false
}

type CustomizedRetryNotMatterActions func(errorCode string, errorMsg string, action string, httpStatus int) bool

func retryNotMatterActions(errorCode string, errorMsg string) bool {
	if errorCode == ROW_OPERATION_CONFLICT || errorCode == NOT_ENOUGH_CAPACITY_UNIT ||
		errorCode == TABLE_NOT_READY || errorCode == PARTITION_UNAVAILABLE ||
		errorCode == SERVER_BUSY || errorCode == STORAGE_SERVER_BUSY || (errorCode == QUOTA_EXHAUSTED && errorMsg == "Too frequent table operations.") {
		return true
	} else {
		return false
	}
}

func isIdempotent(action string) bool {
	return action == batchGetRowUri || action == describeTableUri ||
		action == getRangeUri || action == getRowUri ||
		action == listTableUri || action == listStreamUri ||
		action == getStreamRecordUri || action == describeStreamUri ||
		action == computeSplitsUri || action == parallelScanUri ||
		action == searchUri || action == describeSearchIndexUri ||
		action == listSearchIndexUri
}

func (internalClient *internalClient) doRequest(url string, uri string, body []byte, resp proto.Message, extraInfo ExtraRequestInfo) ([]byte, error, string) {
	hreq, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
	if err != nil {
		return nil, err, ""
	}
	akInfo := internalClient.credentialsProvider.GetCredentials()
	/* set headers */
	hreq.Header.Set("User-Agent", userAgent)

	date := time.Now().UTC().Format(xOtsDateFormat)

	hreq.Header.Set(xOtsDate, date)
	hreq.Header.Set(xOtsApiversion, ApiVersion)
	hreq.Header.Set(xOtsAccesskeyid, akInfo.GetAccessKeyID())
	hreq.Header.Set(xOtsInstanceName, internalClient.instanceName)
	for key, value := range internalClient.externalHeader {
		hreq.Header[key] = []string{value}
	}

	md5Byte := md5.Sum(body)
	md5Base64 := base64.StdEncoding.EncodeToString(md5Byte[:16])
	hreq.Header.Set(xOtsContentmd5, md5Base64)

	otshead := createOtsHeaders(akInfo.GetAccessKeySecret())
	otshead.set(xOtsDate, date)
	otshead.set(xOtsApiversion, ApiVersion)
	otshead.set(xOtsAccesskeyid, akInfo.GetAccessKeyID())

	if extraInfo.userTraceID != nil && *extraInfo.userTraceID != "" {
		hreq.Header.Set(xOtsHeaderSDKTraceID, *extraInfo.userTraceID)
		otshead.set(xOtsHeaderSDKTraceID, *extraInfo.userTraceID)
	}
	if extraInfo.requestExtension != nil && extraInfo.requestExtension.priority != nil {
		hreq.Header.Set(xOtsHeaderRequestPriority, strconv.Itoa(int(*extraInfo.requestExtension.priority)))
		otshead.set(xOtsHeaderRequestPriority, strconv.Itoa(int(*extraInfo.requestExtension.priority)))
	}
	if extraInfo.requestExtension != nil && extraInfo.requestExtension.tag != nil && *extraInfo.requestExtension.tag != "" {
		hreq.Header.Set(xOtsHeaderRequestTag, *extraInfo.requestExtension.tag)
		otshead.set(xOtsHeaderRequestTag, *extraInfo.requestExtension.tag)
	}

	if akInfo.GetSecurityToken() != "" {
		hreq.Header.Set(xOtsHeaderStsToken, akInfo.GetSecurityToken())
		otshead.set(xOtsHeaderStsToken, akInfo.GetSecurityToken())
	}
	otshead.set(xOtsContentmd5, md5Base64)
	otshead.set(xOtsInstanceName, internalClient.instanceName)
	for key, value := range internalClient.externalHeader {
		if strings.HasPrefix(key, xOtsPrefix) {
			otshead.set(key, value)
		}
	}
	sign, err := otshead.signature(uri, "POST", akInfo.GetAccessKeySecret())

	if err != nil {
		return nil, err, ""
	}
	hreq.Header.Set(xOtsSignature, sign)

	/* end set headers */
	return internalClient.postReq(hreq, url)
}

func (tableStoreClient *TableStoreClient) GetExternalHeader() map[string]string {
	return tableStoreClient.externalHeader
}

func (tableStoreClient *TableStoreClient) GetRetryNotify() RetryNotify {
	if tableStoreClient.internalClient == nil {
		return nil
	}
	return tableStoreClient.RetryNotify
}

func (tableStoreClient *TableStoreClient) SetRetryNotify(retryNotify RetryNotify) {
	if tableStoreClient.internalClient == nil {
		return
	}
	tableStoreClient.RetryNotify = retryNotify
}

// table API
// Create a table with the CreateTableRequest, in which the table name and
// primary keys are required.
// 根据CreateTableRequest创建一个表，其中表名和主健列是必选项
//
// @param request of CreateTableRequest.
// @return Void. 无返回值。
func (tableStoreClient *TableStoreClient) CreateTable(request *CreateTableRequest) (*CreateTableResponse, error) {
	if len(request.TableMeta.TableName) > maxTableNameLength {
		return nil, errTableNameTooLong(request.TableMeta.TableName)
	}

	if len(request.TableMeta.SchemaEntry) > maxPrimaryKeyNum {
		return nil, errPrimaryKeyTooMuch
	}

	if len(request.TableMeta.SchemaEntry) == 0 {
		return nil, errCreateTableNoPrimaryKey
	}

	req := new(otsprotocol.CreateTableRequest)
	req.TableMeta = new(otsprotocol.TableMeta)
	req.TableMeta.TableName = proto.String(request.TableMeta.TableName)

	if len(request.TableMeta.DefinedColumns) > 0 {
		for _, value := range request.TableMeta.DefinedColumns {
			req.TableMeta.DefinedColumn = append(req.TableMeta.DefinedColumn, &otsprotocol.DefinedColumnSchema{Name: &value.Name, Type: value.ColumnType.ConvertToPbDefinedColumnType().Enum()})
		}
	}

	if len(request.IndexMetas) > 0 {
		for _, value := range request.IndexMetas {
			req.IndexMetas = append(req.IndexMetas, value.ConvertToPbIndexMeta())
		}
	}

	for _, key := range request.TableMeta.SchemaEntry {
		keyType := otsprotocol.PrimaryKeyType(*key.Type)
		if key.Option != nil {
			keyOption := otsprotocol.PrimaryKeyOption(*key.Option)
			req.TableMeta.PrimaryKey = append(req.TableMeta.PrimaryKey, &otsprotocol.PrimaryKeySchema{Name: key.Name, Type: &keyType, Option: &keyOption})
		} else {
			req.TableMeta.PrimaryKey = append(req.TableMeta.PrimaryKey, &otsprotocol.PrimaryKeySchema{Name: key.Name, Type: &keyType})
		}
	}

	req.ReservedThroughput = new(otsprotocol.ReservedThroughput)
	req.ReservedThroughput.CapacityUnit = new(otsprotocol.CapacityUnit)
	req.ReservedThroughput.CapacityUnit.Read = proto.Int32(int32(request.ReservedThroughput.Readcap))
	req.ReservedThroughput.CapacityUnit.Write = proto.Int32(int32(request.ReservedThroughput.Writecap))

	req.TableOptions = new(otsprotocol.TableOptions)
	req.TableOptions.TimeToLive = proto.Int32(int32(request.TableOption.TimeToAlive))
	req.TableOptions.MaxVersions = proto.Int32(int32(request.TableOption.MaxVersion))
	if request.TableOption.AllowUpdate != nil {
		req.TableOptions.AllowUpdate = proto.Bool(*request.TableOption.AllowUpdate)
	}

	if request.TableOption.DeviationCellVersionInSec > 0 {
		req.TableOptions.DeviationCellVersionInSec = proto.Int64(request.TableOption.DeviationCellVersionInSec)
	}

	if request.StreamSpec != nil {
		var ss otsprotocol.StreamSpecification
		if request.StreamSpec.EnableStream {
			ss = otsprotocol.StreamSpecification{
				EnableStream:   &request.StreamSpec.EnableStream,
				ExpirationTime: &request.StreamSpec.ExpirationTime,
				ColumnsToGet:   request.StreamSpec.OriginColumnsToGet,
			}
		} else {
			ss = otsprotocol.StreamSpecification{
				EnableStream: &request.StreamSpec.EnableStream}
		}

		req.StreamSpec = &ss
	}

	if request.SSESpecification != nil {
		if err := request.SSESpecification.CheckArguments(); err != nil {
			return nil, err
		}
		sse := new(otsprotocol.SSESpecification)
		sse.Enable = proto.Bool(request.SSESpecification.Enable)
		if request.SSESpecification.KeyType != nil {
			sseType := *request.SSESpecification.KeyType
			switch sseType {
			case SSE_KMS_SERVICE:
				keyType := otsprotocol.SSEKeyType_SSE_KMS_SERVICE
				sse.KeyType = &keyType
			case SSE_BYOK:
				keyType := otsprotocol.SSEKeyType_SSE_BYOK
				sse.KeyType = &keyType
			default:
				return nil, errInvalidSSEKeyType(sseType.String())
			}
		}

		if request.SSESpecification.KeyId != nil {
			sse.KeyId = []byte(*request.SSESpecification.KeyId)
		}

		if request.SSESpecification.RoleArn != nil {
			sse.RoleArn = []byte(*request.SSESpecification.RoleArn)
		}

		req.SseSpec = sse
	}

	if request.EnableLocalTxn != nil {
		req.EnableLocalTxn = request.EnableLocalTxn
	}

	resp := new(otsprotocol.CreateTableResponse)
	response := &CreateTableResponse{}
	if err := tableStoreClient.doRequestWithRetry(createTableUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

// Create a timeseries table with CreateTimeseriesTableRequest. in which the timeseriesname
// and tableOptions are required.
// 根据CreateTimeseriesTableRequest创建一个时序表，其中表名和表选项是必选项
//
// @param request of CreateTimeseriesTableRequest。
// @return Void. 无返回值。
func (timeseriesClient *TimeseriesClient) CreateTimeseriesTable(request *CreateTimeseriesTableRequest) (*CreateTimeseriesTableResponse, error) {
	req := new(otsprotocol.CreateTimeseriesTableRequest)
	req.TableMeta = new(otsprotocol.TimeseriesTableMeta)
	req.TableMeta.TableName = proto.String(request.GetTimeseriesTableMeta().GetTimeseriesTableName())

	req.TableMeta.TableOptions = new(otsprotocol.TimeseriesTableOptions)
	req.TableMeta.TableOptions.TimeToLive = proto.Int32(int32(request.GetTimeseriesTableMeta().GetTimeseriesTableOPtions().GetTimeToLive()))

	for _, analyticalStore := range request.GetAnalyticalStores() {
		req.AnalyticalStores = append(req.AnalyticalStores, &otsprotocol.TimeseriesAnalyticalStore{
			StoreName:  proto.String(analyticalStore.StoreName),
			TimeToLive: analyticalStore.TimeToLive,
			SyncOption: (*otsprotocol.AnalyticalStoreSyncType)(analyticalStore.SyncOption),
		})
	}
	req.EnableAnalyticalStore = proto.Bool(request.GetEnableAnalyticalStore())

	resp := new(otsprotocol.CreateTimeseriesTableResponse)
	response := &CreateTimeseriesTableResponse{}
	if err := timeseriesClient.doRequestWithRetry(createTimeseriesTable, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	return response, nil
}

// Put a row in a timeseries table. The timeseriesTableName and TimeseriesRow are required.
// 插入新的时序数据，其中时序表名和时序数据行(可多行)作为参数
//
// @param request of PutTimeseriesDataRequest。
// @return FailedRowResult
func (timeseriesClient *TimeseriesClient) PutTimeseriesData(request *PutTimeseriesDataRequest) (*PutTimeseriesDataResponse, error) {
	if request == nil || request.timeseriesTableName == "" || request.rows == nil || len(request.rows) == 0 {
		return nil, fmt.Errorf("PutTimeseriesDataRequest is empty")
	}

	var err error
	req := new(otsprotocol.PutTimeseriesDataRequest)
	req.TableName = proto.String(request.timeseriesTableName)

	req.RowsData = new(otsprotocol.TimeseriesRows)
	req.RowsData.Type = new(otsprotocol.RowsSerializeType)
	req.RowsData.Type = RowsSerializeType.Enum()
	switch RowsSerializeType {
	case otsprotocol.RowsSerializeType_RST_FLAT_BUFFER:
		req.RowsData.RowsData, err = BuildFlatbufferRows(request.rows, request.timeseriesTableName, timeseriesClient.timeseriesMetaCache)
	case otsprotocol.RowsSerializeType_RST_PROTO_BUFFER:
		req.RowsData.RowsData, err = buildProtocolBufferRows(request.rows, request.timeseriesTableName, timeseriesClient.timeseriesMetaCache)
	default:
		err = fmt.Errorf("Invalid rows serialize type")
	}
	if err != nil {
		return nil, err
	}

	// Compute crc32
	uint32_crc32 := crc32.Checksum(req.RowsData.RowsData, crc32.MakeTable(crc32.Castagnoli))
	req.RowsData.FlatbufferCrc32C = proto.Int32(int32((uint32_crc32)))

	resp := new(otsprotocol.PutTimeseriesDataResponse)
	response := &PutTimeseriesDataResponse{}
	if err := timeseriesClient.doRequestWithRetry(putTimeseriesData, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	if len(resp.FailedRows) != 0 {
		response.failedRowResults = []*FailedRowResult{}
		for _, protoFailedRowResult := range resp.FailedRows {
			var respFailedRowResult FailedRowResult
			respFailedRowResult.Index = *protoFailedRowResult.RowIndex
			respFailedRowResult.Error = fmt.Errorf("ErrorCode: %v, Message: %v", protoFailedRowResult.GetErrorCode(), protoFailedRowResult.GetErrorMessage())
			respFailedRowResult.ErrorCode = protoFailedRowResult.GetErrorCode()
			response.failedRowResults = append(response.failedRowResults, &respFailedRowResult)
		}
	}

	if resp.GetMetaUpdateStatus() != nil && len(resp.GetMetaUpdateStatus().GetRowIds()) != 0 {
		for i := 0; i < len(resp.GetMetaUpdateStatus().GetRowIds()); i++ {
			if i >= len(resp.GetMetaUpdateStatus().GetMetaUpdateTimes()) {
				break
			}
			idx := resp.GetMetaUpdateStatus().GetRowIds()[i]
			if int(idx) < len(request.GetTimeseriesRows()) {
				updateTimeInSec := uint32(resp.GetMetaUpdateStatus().GetMetaUpdateTimes()[i] & 0xffffffff)
				if updateTimeInSec > 0 {
					curRow := request.GetTimeseriesRows()[idx]
					if curRow.timeseriesMetaKey == nil {
						curRow.timeseriesMetaKey = new(string)
						*curRow.timeseriesMetaKey, _ = curRow.timeseriesKey.buildTimeseriesMetaKey(request.GetTimeseriesTableName())
					}
					metaCacheKey := *curRow.timeseriesMetaKey
					timeInCache, ok := timeseriesClient.GetTimeseriesMetaCache().Get(metaCacheKey)
					if !ok || timeInCache.(uint32) < updateTimeInSec {
						timeseriesClient.GetTimeseriesMetaCache().Add(metaCacheKey, updateTimeInSec)
					}
				}
			}
		}
	}
	return response, nil
}

// row API
// Get the timeseries data of a row or some columns.
// 获取某一时间线的一个或多个数据点
//
// @param GetTimeseriesDataRequest
// @return GetTimeseriesDataResponse
func (timeseriesClient *TimeseriesClient) GetTimeseriesData(request *GetTimeseriesDataRequest) (*GetTimeseriesDataResponse, error) {
	if request == nil || request.GetTimeseriesTableName() == "" || request.GetTimeseriesKey() == nil {
		return nil, fmt.Errorf("GetTimeseriesDataRequest is empty")
	}
	if request.GetBeginTimeInUs() >= request.GetEndTimeInUs() {
		return nil, fmt.Errorf("End time should be large than begin time")
	}

	var err error
	req := new(otsprotocol.GetTimeseriesDataRequest)
	req.TableName = proto.String(request.GetTimeseriesTableName())
	req.BeginTime = proto.Int64(request.GetBeginTimeInUs())
	req.EndTime = proto.Int64(request.GetEndTimeInUs())
	req.TimeSeriesKey, err = buildTimeseriesKey(request.GetTimeseriesKey())
	if err != nil {
		return nil, err
	}

	if request.GetLimit() > 0 {
		req.Limit = proto.Int32(int32(request.GetLimit()))
	}

	if request.GetNextToken() != nil && len(request.GetNextToken()) != 0 {
		req.Token = append([]byte{}, request.GetNextToken()...)
	}

	if request.backward {
		req.Backward = proto.Bool(true)
	}

	if len(request.GetFieldsToGet()) > 0 {
		fieldsToGet := request.GetFieldsToGet()
		req.FieldsToGet = make([]*otsprotocol.TimeseriesFieldsToGet, 0, len(fieldsToGet))
		for i, field := range fieldsToGet {
			var fieldType int32 = -1
			switch field.Type {
			case ColumnType_INTEGER:
				fieldType = int32(Fieldvalues.DataTypeLONG)
			case ColumnType_BOOLEAN:
				fieldType = int32(Fieldvalues.DataTypeBOOLEAN)
			case ColumnType_DOUBLE:
				fieldType = int32(Fieldvalues.DataTypeDOUBLE)
			case ColumnType_STRING:
				fieldType = int32(Fieldvalues.DataTypeSTRING)
			case ColumnType_BINARY:
				fieldType = int32(Fieldvalues.DataTypeBINARY)
			default:
				return nil, errors.New(fmt.Sprintf("invalid type: %v", field.Type))
			}
			req.FieldsToGet = append(req.FieldsToGet, &otsprotocol.TimeseriesFieldsToGet{
				Name: &fieldsToGet[i].Name,
				Type: proto.Int32(fieldType),
			})
		}
	}

	resp := new(otsprotocol.GetTimeseriesDataResponse)
	response := new(GetTimeseriesDataResponse)

	if err = timeseriesClient.doRequestWithRetry(getTimeseriesData, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	builderResponse, err := CreateGetTimeseriesDataResponse(resp)
	if err != nil {
		return nil, err
	}
	response.rows = builderResponse.rows
	response.nextToken = builderResponse.nextToken
	return response, nil
}

// Get timeseries table meta infomation
// 获取指定时序表的元数据
//
// @param request of DescribeTimeseriesTableRequest.
// @return TimeseriesTableMeta
func (timeseriesClient *TimeseriesClient) DescribeTimeseriesTable(request *DescribeTimeseriesTableRequest) (*DescribeTimeseriesTableResponse, error) {
	if request.GetTimeseriesTableName() == "" {
		return nil, fmt.Errorf("DescribeTimeseriesTableRequest.timeseriesTableName is empty")
	}

	req := new(otsprotocol.DescribeTimeseriesTableRequest)
	req.TableName = proto.String(request.GetTimeseriesTableName())

	resp := new(otsprotocol.DescribeTimeseriesTableResponse)
	response := new(DescribeTimeseriesTableResponse)
	if err := timeseriesClient.doRequestWithRetry(describeTimeseriesTable, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	if resp.GetTableMeta() != nil && resp.GetTableMeta().GetTableName() != "" {
		response.timeseriesTableMeta = ParseTimeseriesTableMeta(resp.GetTableMeta())
	}
	for _, analyticalStore := range resp.GetAnalyticalStores() {
		response.analyticalStores = append(response.analyticalStores, TimeseriesAnalyticalStore{
			StoreName:  analyticalStore.GetStoreName(),
			TimeToLive: analyticalStore.TimeToLive,
			SyncOption: (*AnalyticalStoreSyncType)(analyticalStore.SyncOption),
		})
	}
	return response, nil
}

// List all timeseries table name in this instance
// 列出该实例中的所有时序表的元数据信息
//
// @param Void
// @return []*TimeseriesTableMeta
func (timeseriesClient *TimeseriesClient) ListTimeseriesTable() (*ListTimeseriesTableResponse, error) {
	req := new(otsprotocol.ListTimeseriesTableRequest)

	resp := new(otsprotocol.ListTimeseriesTableResponse)
	response := new(ListTimeseriesTableResponse)

	if err := timeseriesClient.doRequestWithRetry(listTimeseriesTable, req, resp, &response.ResponseInfo, ExtraRequestInfo{}); err != nil {
		return nil, err
	}

	for i := 0; i < len(resp.GetTableMetas()); i++ {
		if resp.GetTableMetas()[i].GetTableOptions() != nil && resp.GetTableMetas()[i].GetTableName() != "" {
			timeseriesTableMeta := ParseTimeseriesTableMeta(resp.GetTableMetas()[i])
			response.timeseriesTableMetas = append(response.timeseriesTableMetas, timeseriesTableMeta)
		}
	}
	return response, nil
}

// Delete a timeseries table
// 删除一个时序表
//
// @param DeleteTimeseriesTableRequest
// return Void
func (timeseriesClient *TimeseriesClient) DeleteTimeseriesTable(request *DeleteTimeseriesTableRequest) (*DeleteTimeseriesTableResponse, error) {
	if request.timeseriesTableName == "" {
		return nil, fmt.Errorf("DeleteTimeseriesTableRequest is empty")
	}

	req := new(otsprotocol.DeleteTimeseriesTableRequest)
	req.TableName = proto.String(request.GetTimeseriesTableName())

	resp := new(otsprotocol.DeleteTimeseriesTableResponse)
	response := new(DeleteTimeseriesTableResponse)

	if err := timeseriesClient.doRequestWithRetry(deleteTimeseriesTable, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	return response, nil
}

// Query timeseries meta(measurement,tag,source) information in a timeseries table.
// 查询一个时序表中的时序元数据(measurement，tag，source)信息。
//
// @param request of QueryTimeseriesMetaRequest
// @return meta information of one or more timeline: QueryTimeseriesMetaResponse
func (timeseriesClient *TimeseriesClient) QueryTimeseriesMeta(request *QueryTimeseriesMetaRequest) (*QueryTimeseriesMetaResponse, error) {
	if request.GetTimeseriesTableName() == "" {
		return nil, fmt.Errorf("QueryTimeseriesMetaRequest is empty")
	}

	req := new(otsprotocol.QueryTimeseriesMetaRequest)

	if request.GetNextToken() != nil && len(request.GetNextToken()) != 0 {
		req.Token = request.GetNextToken()
	}

	req.TableName = proto.String(request.timeseriesTableName)
	req.GetTotalHit = proto.Bool(request.getTotalHits)

	if request.GetCondition() != nil {
		reqCondition := new(otsprotocol.MetaQueryCondition)
		reqCondition.Type = otsprotocol.MetaQueryConditionType(int32((request.GetCondition()).GetType())).Enum()
		reqCondition.ProtoData = request.GetCondition().Serialize()
		req.Condition = reqCondition
	}

	if request.GetLimit() > 0 {
		req.Limit = proto.Int32(int32(request.GetLimit()))
	}

	resp := new(otsprotocol.QueryTimeseriesMetaResponse)
	response := new(QueryTimeseriesMetaResponse)
	if err := timeseriesClient.doRequestWithRetry(queryTimeseriesMeta, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.nextToken = resp.GetNextToken()
	response.totalHits = resp.GetTotalHit()
	for i := 0; i < len(resp.GetTimeseriesMetas()); i++ {
		currentRespTimeseriesMeta := resp.GetTimeseriesMetas()[i]
		timeseriesMeta, err := parseTimeseriesMeta(currentRespTimeseriesMeta)
		if err != nil {
			return nil, err
		}
		response.timeseriesMetas = append(response.timeseriesMetas, timeseriesMeta)
	}
	return response, nil
}

// update timeInus parameter for a timeseries table.
// 更新一个时序表的TTL参数
//
// @param UpdateTimeseriesTableRequest
// @return Void
func (timeseriesClient *TimeseriesClient) UpdateTimeseriesTable(request *UpdateTimeseriesTableRequest) (*UpdateTimeseriesTableResponse, error) {
	if request.GetTimeseriesTableName() == "" || request.GetTimeseriesTableOptions() == nil {
		return nil, fmt.Errorf("UpdateTimeseriesTableRequest is empty")
	}

	req := new(otsprotocol.UpdateTimeseriesTableRequest)
	req.TableName = proto.String(request.GetTimeseriesTableName())
	reqTimeseriesTableOptions := new(otsprotocol.TimeseriesTableOptions)
	reqTimeseriesTableOptions.TimeToLive = proto.Int32(int32(request.GetTimeseriesTableOptions().GetTimeToLive()))
	req.TableOptions = reqTimeseriesTableOptions

	resp := new(otsprotocol.UpdateTimeseriesTableRequest)
	response := new(UpdateTimeseriesTableResponse)

	if err := timeseriesClient.doRequestWithRetry(updateTimeseriesTable, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

// update timeseries attributes for time line.
// 更新时间线的属性信息
//
// @param UpdateTimeseriesMetaRequest
// @return UpdateTimeseriesMetaResponse
func (timeseriesClient *TimeseriesClient) UpdateTimeseriesMeta(request *UpdateTimeseriesMetaRequest) (*UpdateTimeseriesMetaResponse, error) {
	if request.GetTimeseriesTableName() == "" {
		return nil, fmt.Errorf("not set timeseries table name")
	}
	if len(request.GetTimeseriesMetas()) == 0 {
		return nil, fmt.Errorf("UpdateTimeseriesMetaRequest is empty")
	}
	var err error

	req := new(otsprotocol.UpdateTimeseriesMetaRequest)
	req.TableName = proto.String(request.GetTimeseriesTableName())
	req.TimeseriesMeta = make([]*otsprotocol.TimeseriesMeta, 0, len(request.GetTimeseriesMetas()))
	for i := 0; i < len(request.GetTimeseriesMetas()); i++ {
		curTimeseriesMeta := request.GetTimeseriesMetas()[i]

		if curTimeseriesMeta.GetUpdateTimeInUs() > 0 {
			return nil, fmt.Errorf("Update time can not be set")
		}

		timeseriesMeta := new(otsprotocol.TimeseriesMeta)
		timeseriesMeta.TimeSeriesKey, err = buildTimeseriesKey(curTimeseriesMeta.GetTimeseriesKey())
		if err != nil {
			return nil, err
		}

		if curTimeseriesMeta.GetAttributes() != nil {
			Attribute, err := BuildTagString(curTimeseriesMeta.GetAttributes())
			if err != nil {
				return nil, err
			}
			timeseriesMeta.Attributes = proto.String(Attribute)
		}

		req.TimeseriesMeta = append(req.TimeseriesMeta, timeseriesMeta)
	}

	resp := new(otsprotocol.UpdateTimeseriesMetaResponse)
	response := new(UpdateTimeseriesMetaResponse)

	if err := timeseriesClient.doRequestWithRetry(updateTimeseriesMeta, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	if len(resp.GetFailedRows()) != 0 {
		response.failedRowResults = []*FailedRowResult{}
		for _, respFailedResult := range resp.GetFailedRows() {
			failedRowResult := &FailedRowResult{}
			failedRowResult.Index = *respFailedResult.RowIndex
			failedRowResult.Error = fmt.Errorf("ErrorCode: %v, Message: %v", respFailedResult.GetErrorCode(), respFailedResult.GetErrorMessage())
			failedRowResult.ErrorCode = respFailedResult.GetErrorCode()
			response.failedRowResults = append(response.failedRowResults, failedRowResult)
		}
	}

	return response, nil
}

// delete timeseries meta
// 删除时间线元数据
//
// @param DeleteTimeseriesMetaRequest
// @return DeleteTimeseriesMetaResponse
func (timeseriesClient *TimeseriesClient) DeleteTimeseriesMeta(request *DeleteTimeseriesMetaRequest) (*DeleteTimeseriesMetaResponse, error) {
	if request.GetTimeseriesTableName() == "" {
		return nil, fmt.Errorf("not set timeseries table name")
	}
	if len(request.GetTimeseriesKeys()) == 0 {
		return nil, fmt.Errorf("DeleteTimeseriesMetaRequest is empty")
	}

	var err error

	req := new(otsprotocol.DeleteTimeseriesMetaRequest)
	req.TableName = proto.String(request.GetTimeseriesTableName())
	req.TimeseriesKey = make([]*otsprotocol.TimeseriesKey, 0, len(request.GetTimeseriesKeys()))
	for i := 0; i < len(request.GetTimeseriesKeys()); i++ {
		timeseriesKey := new(otsprotocol.TimeseriesKey)
		timeseriesKey, err = buildTimeseriesKey(request.GetTimeseriesKeys()[i])
		if err != nil {
			return nil, err
		}
		req.TimeseriesKey = append(req.TimeseriesKey, timeseriesKey)
	}

	resp := new(otsprotocol.DeleteTimeseriesMetaResponse)
	response := new(DeleteTimeseriesMetaResponse)

	if err := timeseriesClient.doRequestWithRetry(deleteTimeseriesMeta, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	if len(resp.GetFailedRows()) != 0 {
		response.failedRowResults = []*FailedRowResult{}
		for _, respFailedResult := range resp.GetFailedRows() {
			failedRowResult := &FailedRowResult{}
			failedRowResult.Index = *respFailedResult.RowIndex
			failedRowResult.Error = fmt.Errorf("ErrorCode: %v, Message: %v", respFailedResult.GetErrorCode(), respFailedResult.GetErrorMessage())
			failedRowResult.ErrorCode = respFailedResult.GetErrorCode()
			response.failedRowResults = append(response.failedRowResults, failedRowResult)
		}
	}

	return response, nil
}

func (timeseriesClient *TimeseriesClient) CreateTimeseriesAnalyticalStore(request *CreateTimeseriesAnalyticalStoreRequest) (*CreateTimeseriesAnalyticalStoreResponse, error) {
	if request.timeseriesTableName == "" {
		return nil, fmt.Errorf("not set timeseries table name")
	}
	if request.analyticalStore.StoreName == "" {
		return nil, fmt.Errorf("not set analytical store name")
	}

	req := new(otsprotocol.CreateTimeseriesAnalyticalStoreRequest)
	req.TableName = proto.String(request.timeseriesTableName)
	req.AnalyticalStore = &otsprotocol.TimeseriesAnalyticalStore{
		StoreName:  proto.String(request.analyticalStore.StoreName),
		TimeToLive: request.analyticalStore.TimeToLive,
		SyncOption: (*otsprotocol.AnalyticalStoreSyncType)(request.analyticalStore.SyncOption),
	}

	resp := new(otsprotocol.CreateTimeseriesAnalyticalStoreResponse)
	response := new(CreateTimeseriesAnalyticalStoreResponse)

	if err := timeseriesClient.doRequestWithRetry(createTimeseriesAnalyticalStore, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

func (timeseriesClient *TimeseriesClient) DeleteTimeseriesAnalyticalStore(request *DeleteTimeseriesAnalyticalStoreRequest) (*DeleteTimeseriesAnalyticalStoreResponse, error) {
	if request.timeseriesTableName == "" {
		return nil, fmt.Errorf("not set timeseries table name")
	}
	if request.analyticalStoreName == "" {
		return nil, fmt.Errorf("not set analytical store name")
	}

	req := new(otsprotocol.DeleteTimeseriesAnalyticalStoreRequest)
	req.TableName = proto.String(request.timeseriesTableName)
	req.StoreName = proto.String(request.analyticalStoreName)
	req.DropMappingTable = proto.Bool(request.dropMappingTable)

	resp := new(otsprotocol.DeleteTimeseriesAnalyticalStoreResponse)
	response := new(DeleteTimeseriesAnalyticalStoreResponse)

	if err := timeseriesClient.doRequestWithRetry(deleteTimeseriesAnalyticalStore, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

func (timeseriesClient *TimeseriesClient) DescribeTimeseriesAnalyticalStore(request *DescribeTimeseriesAnalyticalStoreRequest) (*DescribeTimeseriesAnalyticalStoreResponse, error) {
	if request.timeseriesTableName == "" {
		return nil, fmt.Errorf("not set timeseries table name")
	}
	if request.analyticalStoreName == "" {
		return nil, fmt.Errorf("not set analytical store name")
	}

	req := new(otsprotocol.DescribeTimeseriesAnalyticalStoreRequest)
	req.TableName = proto.String(request.timeseriesTableName)
	req.StoreName = proto.String(request.analyticalStoreName)

	resp := new(otsprotocol.DescribeTimeseriesAnalyticalStoreResponse)
	response := new(DescribeTimeseriesAnalyticalStoreResponse)

	if err := timeseriesClient.doRequestWithRetry(describeTimeseriesAnalyticalStore, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.AnalyticalStore = &TimeseriesAnalyticalStore{
		StoreName:  resp.GetAnalyticalStore().GetStoreName(),
		TimeToLive: resp.GetAnalyticalStore().TimeToLive,
		SyncOption: (*AnalyticalStoreSyncType)(resp.GetAnalyticalStore().SyncOption),
	}
	if resp.StorageSize != nil {
		response.StorageSize = &AnalyticalStoreStorageSize{
			Size:      resp.GetStorageSize().GetSize(),
			Timestamp: resp.GetStorageSize().GetTimestamp(),
		}
	}
	if resp.SyncStat != nil {
		response.SyncStat = &AnalyticalStoreSyncStat{
			CurrentSyncTimestamp: resp.GetSyncStat().GetCurrentSyncTimestamp(),
			SyncPhase:            (AnalyticalStoreSyncType)(resp.GetSyncStat().GetSyncPhase()),
		}
	}

	return response, nil
}

func (timeseriesClient *TimeseriesClient) UpdateTimeseriesAnalyticalStore(request *UpdateTimeseriesAnalyticalStoreRequest) (*UpdateTimeseriesAnalyticalStoreResponse, error) {
	if request.timeseriesTableName == "" {
		return nil, fmt.Errorf("not set timeseries table name")
	}
	if request.analyticalStore.StoreName == "" {
		return nil, fmt.Errorf("not set analytical store name")
	}

	req := new(otsprotocol.UpdateTimeseriesAnalyticalStoreRequest)
	req.TableName = proto.String(request.timeseriesTableName)
	req.AnalyticalStore = &otsprotocol.TimeseriesAnalyticalStore{
		StoreName:  proto.String(request.analyticalStore.StoreName),
		TimeToLive: request.analyticalStore.TimeToLive,
		SyncOption: (*otsprotocol.AnalyticalStoreSyncType)(request.analyticalStore.SyncOption),
	}

	resp := new(otsprotocol.UpdateTimeseriesAnalyticalStoreResponse)
	response := new(UpdateTimeseriesAnalyticalStoreResponse)

	if err := timeseriesClient.doRequestWithRetry(updateTimeseriesAnalyticalStore, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

func (tableStoreClient *TableStoreClient) CreateIndex(request *CreateIndexRequest) (*CreateIndexResponse, error) {
	if len(request.MainTableName) > maxTableNameLength {
		return nil, errTableNameTooLong(request.MainTableName)
	}

	req := new(otsprotocol.CreateIndexRequest)
	req.IndexMeta = request.IndexMeta.ConvertToPbIndexMeta()
	req.IncludeBaseData = proto.Bool(request.IncludeBaseData)
	req.MainTableName = proto.String(request.MainTableName)

	resp := new(otsprotocol.CreateIndexResponse)
	response := &CreateIndexResponse{}
	if err := tableStoreClient.doRequestWithRetry(createIndexUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

func (tableStoreClient *TableStoreClient) DeleteIndex(request *DeleteIndexRequest) (*DeleteIndexResponse, error) {
	if len(request.MainTableName) > maxTableNameLength {
		return nil, errTableNameTooLong(request.MainTableName)
	}

	req := new(otsprotocol.DropIndexRequest)
	req.IndexName = proto.String(request.IndexName)
	req.MainTableName = proto.String(request.MainTableName)

	resp := new(otsprotocol.DropIndexResponse)
	response := &DeleteIndexResponse{}
	if err := tableStoreClient.doRequestWithRetry(dropIndexUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

// List all tables. If done, all table names will be returned.
// 列出所有的表，如果操作成功，将返回所有表的名称。
//
// @param tableNames The returned table names. 返回的表名集合。
// @return Void. 无返回值。
func (tableStoreClient *TableStoreClient) ListTable() (*ListTableResponse, error) {
	resp := new(otsprotocol.ListTableResponse)
	response := &ListTableResponse{}
	if err := tableStoreClient.doRequestWithRetry(listTableUri, nil, resp, &response.ResponseInfo, ExtraRequestInfo{}); err != nil {
		return response, err
	}

	response.TableNames = resp.TableNames
	return response, nil
}

// Delete a table and all its views will be deleted.
// 删除一个表
//
// @param tableName The table name. 表名。
// @return Void. 无返回值。
func (tableStoreClient *TableStoreClient) DeleteTable(request *DeleteTableRequest) (*DeleteTableResponse, error) {
	req := new(otsprotocol.DeleteTableRequest)
	req.TableName = proto.String(request.TableName)

	response := &DeleteTableResponse{}
	if err := tableStoreClient.doRequestWithRetry(deleteTableUri, req, nil, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	return response, nil
}

// Query the tablemeta, tableoption and reservedthroughtputdetails
// @param DescribeTableRequest
// @param DescribeTableResponse
func (tableStoreClient *TableStoreClient) DescribeTable(request *DescribeTableRequest) (*DescribeTableResponse, error) {
	req := new(otsprotocol.DescribeTableRequest)
	req.TableName = proto.String(request.TableName)

	resp := new(otsprotocol.DescribeTableResponse)
	response := new(DescribeTableResponse)

	if err := tableStoreClient.doRequestWithRetry(describeTableUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return &DescribeTableResponse{}, err
	}

	response.ReservedThroughput = &ReservedThroughput{Readcap: int(*(resp.ReservedThroughputDetails.CapacityUnit.Read)), Writecap: int(*(resp.ReservedThroughputDetails.CapacityUnit.Write))}

	responseTableMeta := new(TableMeta)
	responseTableMeta.TableName = *resp.TableMeta.TableName

	for _, key := range resp.TableMeta.PrimaryKey {
		keyType := PrimaryKeyType(*key.Type)

		// enable it when we support kep option in describe table
		if key.Option != nil {
			keyOption := PrimaryKeyOption(*key.Option)
			responseTableMeta.SchemaEntry = append(responseTableMeta.SchemaEntry, &PrimaryKeySchema{Name: key.Name, Type: &keyType, Option: &keyOption})
		} else {
			responseTableMeta.SchemaEntry = append(responseTableMeta.SchemaEntry, &PrimaryKeySchema{Name: key.Name, Type: &keyType})
		}
	}

	for _, value := range resp.TableMeta.DefinedColumn {
		definedColumn := &DefinedColumnSchema{*value.Name, ConvertPbDefinedColumnType(*value.Type)}
		responseTableMeta.DefinedColumns = append(responseTableMeta.DefinedColumns, definedColumn)
	}

	response.TableMeta = responseTableMeta
	allowUpdate := true
	if resp.TableOptions.AllowUpdate != nil {
		allowUpdate = *resp.TableOptions.AllowUpdate
	}

	response.TableOption = &TableOption{
		TimeToAlive:               int(*resp.TableOptions.TimeToLive),
		MaxVersion:                int(*resp.TableOptions.MaxVersions),
		DeviationCellVersionInSec: *resp.TableOptions.DeviationCellVersionInSec,
		AllowUpdate:               &allowUpdate,
	}

	if resp.StreamDetails != nil && *resp.StreamDetails.EnableStream {
		response.StreamDetails = &StreamDetails{
			EnableStream:       *resp.StreamDetails.EnableStream,
			StreamId:           (*StreamId)(resp.StreamDetails.StreamId),
			ExpirationTime:     *resp.StreamDetails.ExpirationTime,
			LastEnableTime:     *resp.StreamDetails.LastEnableTime,
			OriginColumnsToGet: resp.GetStreamDetails().GetColumnsToGet(),
		}
	} else {
		response.StreamDetails = &StreamDetails{
			EnableStream: false}
	}

	for _, meta := range resp.IndexMetas {
		response.IndexMetas = append(response.IndexMetas, ConvertPbIndexMetaToIndexMeta(meta))
	}

	if resp.GetSseDetails() == nil {
		response.SSEDetails = &SSEDetails{
			Enable: false,
		}
	} else {
		respSse := resp.GetSseDetails()
		sseDetail := new(SSEDetails)
		sseDetail.Enable = resp.GetSseDetails().GetEnable()
		switch respSse.GetKeyType() {
		case otsprotocol.SSEKeyType_SSE_KMS_SERVICE:
			sseDetail.KeyType = SSE_KMS_SERVICE
		case otsprotocol.SSEKeyType_SSE_BYOK:
			sseDetail.KeyType = SSE_BYOK
		default:
			return nil, errInvalidSSEKeyType(respSse.GetKeyType().String())
		}

		if respSse.GetKeyId() != nil {
			sseDetail.KeyId = string(respSse.GetKeyId())
		}

		if respSse.GetRoleArn() != nil {
			sseDetail.RoleArn = string(respSse.GetRoleArn())
		}

		response.SSEDetails = sseDetail
	}

	return response, nil
}

// Update the table info includes tableoptions and reservedthroughput
// @param UpdateTableRequest
// @param UpdateTableResponse
func (tableStoreClient *TableStoreClient) UpdateTable(request *UpdateTableRequest) (*UpdateTableResponse, error) {
	req := new(otsprotocol.UpdateTableRequest)
	req.TableName = proto.String(request.TableName)

	if request.ReservedThroughput != nil {
		req.ReservedThroughput = new(otsprotocol.ReservedThroughput)
		req.ReservedThroughput.CapacityUnit = new(otsprotocol.CapacityUnit)
		req.ReservedThroughput.CapacityUnit.Read = proto.Int32(int32(request.ReservedThroughput.Readcap))
		req.ReservedThroughput.CapacityUnit.Write = proto.Int32(int32(request.ReservedThroughput.Writecap))
	}

	if request.TableOption != nil {
		req.TableOptions = new(otsprotocol.TableOptions)
		req.TableOptions.TimeToLive = proto.Int32(int32(request.TableOption.TimeToAlive))
		req.TableOptions.MaxVersions = proto.Int32(int32(request.TableOption.MaxVersion))
		if request.TableOption.AllowUpdate != nil {
			req.TableOptions.AllowUpdate = proto.Bool(*request.TableOption.AllowUpdate)
		}

		if request.TableOption.DeviationCellVersionInSec > 0 {
			req.TableOptions.DeviationCellVersionInSec = proto.Int64(request.TableOption.DeviationCellVersionInSec)
		}
	}

	if request.StreamSpec != nil {
		if request.StreamSpec.EnableStream == true {
			req.StreamSpec = &otsprotocol.StreamSpecification{
				EnableStream:   &request.StreamSpec.EnableStream,
				ExpirationTime: &request.StreamSpec.ExpirationTime,
				ColumnsToGet:   request.StreamSpec.OriginColumnsToGet,
			}
		} else {
			req.StreamSpec = &otsprotocol.StreamSpecification{EnableStream: &request.StreamSpec.EnableStream}
		}
	}

	resp := new(otsprotocol.UpdateTableResponse)
	response := new(UpdateTableResponse)

	if err := tableStoreClient.doRequestWithRetry(updateTableUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.ReservedThroughput = &ReservedThroughput{
		Readcap:  int(*(resp.ReservedThroughputDetails.CapacityUnit.Read)),
		Writecap: int(*(resp.ReservedThroughputDetails.CapacityUnit.Write))}
	response.TableOption = &TableOption{
		TimeToAlive:               int(*resp.TableOptions.TimeToLive),
		MaxVersion:                int(*resp.TableOptions.MaxVersions),
		DeviationCellVersionInSec: *resp.TableOptions.DeviationCellVersionInSec}

	if *resp.StreamDetails.EnableStream {
		response.StreamDetails = &StreamDetails{
			EnableStream:       *resp.StreamDetails.EnableStream,
			StreamId:           (*StreamId)(resp.StreamDetails.StreamId),
			ExpirationTime:     *resp.StreamDetails.ExpirationTime,
			LastEnableTime:     *resp.StreamDetails.LastEnableTime,
			OriginColumnsToGet: resp.GetStreamDetails().GetColumnsToGet(),
		}
	} else {
		response.StreamDetails = &StreamDetails{
			EnableStream: false}
	}
	return response, nil
}

func (tableStoreClient *TableStoreClient) AddDefinedColumn(request *AddDefinedColumnRequest) (*AddDefinedColumnResponse, error) {
	req := new(otsprotocol.AddDefinedColumnRequest)
	req.TableName = proto.String(request.TableName)

	if len(request.DefinedColumns) > 0 {
		for _, value := range request.DefinedColumns {
			req.Columns = append(req.Columns, &otsprotocol.DefinedColumnSchema{Name: &value.Name, Type: value.ColumnType.ConvertToPbDefinedColumnType().Enum()})
		}
	}

	resp := new(otsprotocol.AddDefinedColumnResponse)
	response := &AddDefinedColumnResponse{}
	if err := tableStoreClient.doRequestWithRetry(adddefinedcolumnuri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	return response, nil
}

func (tableStoreClient *TableStoreClient) DeleteDefinedColumn(request *DeleteDefinedColumnRequest) (*DeleteDefinedColumnResponse, error) {
	req := new(otsprotocol.DeleteDefinedColumnRequest)
	req.TableName = proto.String(request.TableName)

	if len(request.DefinedColumns) > 0 {
		for _, value := range request.DefinedColumns {
			req.Columns = append(req.Columns, value)
		}
	}

	resp := new(otsprotocol.DeleteDefinedColumnResponse)
	response := &DeleteDefinedColumnResponse{}
	if err := tableStoreClient.doRequestWithRetry(deletedefinedcolumnuri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	return response, nil
}

// Put or update a row in a table. The operation is determined by CheckingType,
// which has three options: NO, UPDATE, INSERT. The transaction id is optional.
// 插入或更新行数据。操作针对数据的存在性包含三种检查类型：NO(不检查)，UPDATE
// （更新，数据必须存在）和INSERT（插入，数据必须不存在）。事务ID是可选项。
//
// @param builder The builder for putting a row. 插入或更新数据的Builder。
// @return Void. 无返回值。
func (tableStoreClient *TableStoreClient) PutRow(request *PutRowRequest) (*PutRowResponse, error) {
	if request == nil {
		return nil, nil
	}

	if request.PutRowChange == nil {
		return nil, nil
	}

	req := new(otsprotocol.PutRowRequest)
	req.TableName = proto.String(request.PutRowChange.TableName)
	req.Row = request.PutRowChange.Serialize()

	condition := new(otsprotocol.Condition)
	condition.RowExistence = request.PutRowChange.Condition.buildCondition()
	if request.PutRowChange.Condition.ColumnCondition != nil {
		condition.ColumnCondition = request.PutRowChange.Condition.ColumnCondition.Serialize()
	}

	if request.PutRowChange.ReturnType == ReturnType_RT_PK {
		content := otsprotocol.ReturnContent{ReturnType: otsprotocol.ReturnType_RT_PK.Enum()}
		req.ReturnContent = &content
	}

	if request.PutRowChange.TransactionId != nil {
		req.TransactionId = request.PutRowChange.TransactionId
	}

	req.Condition = condition

	resp := new(otsprotocol.PutRowResponse)
	response := &PutRowResponse{}
	if err := tableStoreClient.doRequestWithRetry(putRowUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.ConsumedCapacityUnit = &ConsumedCapacityUnit{}
	response.ConsumedCapacityUnit.Read = *resp.Consumed.CapacityUnit.Read
	response.ConsumedCapacityUnit.Write = *resp.Consumed.CapacityUnit.Write

	if request.PutRowChange.ReturnType == ReturnType_RT_PK {
		rows, err := readRowsWithHeader(bytes.NewReader(resp.Row))
		if err != nil {
			return response, err
		}

		for _, pk := range rows[0].primaryKey {
			pkColumn := &PrimaryKeyColumn{ColumnName: string(pk.cellName), Value: pk.cellValue.Value}
			response.PrimaryKey.PrimaryKeys = append(response.PrimaryKey.PrimaryKeys, pkColumn)
		}
	}

	return response, nil
}

// Delete row with pk
// @param DeleteRowRequest
func (tableStoreClient *TableStoreClient) DeleteRow(request *DeleteRowRequest) (*DeleteRowResponse, error) {
	req := new(otsprotocol.DeleteRowRequest)
	req.TableName = proto.String(request.DeleteRowChange.TableName)
	req.Condition = request.DeleteRowChange.getCondition()
	req.PrimaryKey = request.DeleteRowChange.PrimaryKey.Build(true)

	if request.DeleteRowChange.TransactionId != nil {
		req.TransactionId = request.DeleteRowChange.TransactionId
	}

	resp := new(otsprotocol.DeleteRowResponse)
	response := &DeleteRowResponse{}
	if err := tableStoreClient.doRequestWithRetry(deleteRowUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.ConsumedCapacityUnit = &ConsumedCapacityUnit{}
	response.ConsumedCapacityUnit.Read = *resp.Consumed.CapacityUnit.Read
	response.ConsumedCapacityUnit.Write = *resp.Consumed.CapacityUnit.Write
	return response, nil
}

// row API
// Get the data of a row or some columns.
//
// @param getrowrequest
func (tableStoreClient *TableStoreClient) GetRow(request *GetRowRequest) (*GetRowResponse, error) {
	req := new(otsprotocol.GetRowRequest)
	resp := new(otsprotocol.GetRowResponse)

	req.TableName = proto.String(request.SingleRowQueryCriteria.TableName)

	if (request.SingleRowQueryCriteria.getColumnsToGet() != nil) && len(request.SingleRowQueryCriteria.getColumnsToGet()) > 0 {
		req.ColumnsToGet = request.SingleRowQueryCriteria.getColumnsToGet()
	}

	req.PrimaryKey = request.SingleRowQueryCriteria.PrimaryKey.Build(false)

	if request.SingleRowQueryCriteria.MaxVersion != 0 {
		req.MaxVersions = proto.Int32(int32(request.SingleRowQueryCriteria.MaxVersion))
	}

	if request.SingleRowQueryCriteria.TransactionId != nil {
		req.TransactionId = request.SingleRowQueryCriteria.TransactionId
	}

	if request.SingleRowQueryCriteria.StartColumn != nil {
		req.StartColumn = request.SingleRowQueryCriteria.StartColumn
	}

	if request.SingleRowQueryCriteria.EndColumn != nil {
		req.EndColumn = request.SingleRowQueryCriteria.EndColumn
	}

	if request.SingleRowQueryCriteria.TimeRange != nil {
		if request.SingleRowQueryCriteria.TimeRange.Specific != 0 {
			req.TimeRange = &otsprotocol.TimeRange{SpecificTime: proto.Int64(request.SingleRowQueryCriteria.TimeRange.Specific)}
		} else {
			req.TimeRange = &otsprotocol.TimeRange{StartTime: proto.Int64(request.SingleRowQueryCriteria.TimeRange.Start), EndTime: proto.Int64(request.SingleRowQueryCriteria.TimeRange.End)}
		}
	} else if request.SingleRowQueryCriteria.MaxVersion == 0 {
		return nil, errInvalidInput
	}

	if request.SingleRowQueryCriteria.Filter != nil {
		req.Filter = request.SingleRowQueryCriteria.Filter.Serialize()
	}

	response := &GetRowResponse{ConsumedCapacityUnit: &ConsumedCapacityUnit{}}
	if err := tableStoreClient.doRequestWithRetry(getRowUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.ConsumedCapacityUnit.Read = *resp.Consumed.CapacityUnit.Read
	response.ConsumedCapacityUnit.Write = *resp.Consumed.CapacityUnit.Write

	if len(resp.Row) == 0 {
		return response, nil
	}

	rows, err := readRowsWithHeader(bytes.NewReader(resp.Row))
	if err != nil {
		return nil, err
	}

	for _, pk := range rows[0].primaryKey {
		pkColumn := &PrimaryKeyColumn{ColumnName: string(pk.cellName), Value: pk.cellValue.Value}
		response.PrimaryKey.PrimaryKeys = append(response.PrimaryKey.PrimaryKeys, pkColumn)
	}

	for _, cell := range rows[0].cells {
		dataColumn := &AttributeColumn{ColumnName: string(cell.cellName), Value: cell.cellValue.Value, Timestamp: cell.cellTimestamp}
		response.Columns = append(response.Columns, dataColumn)
	}

	return response, nil
}

// Update row
// @param UpdateRowRequest
func (tableStoreClient *TableStoreClient) UpdateRow(request *UpdateRowRequest) (*UpdateRowResponse, error) {
	req := new(otsprotocol.UpdateRowRequest)
	resp := new(otsprotocol.UpdateRowResponse)

	req.TableName = proto.String(request.UpdateRowChange.TableName)
	req.Condition = request.UpdateRowChange.getCondition()
	req.RowChange = request.UpdateRowChange.Serialize()
	if request.UpdateRowChange.TransactionId != nil {
		req.TransactionId = request.UpdateRowChange.TransactionId
	}

	response := &UpdateRowResponse{ConsumedCapacityUnit: &ConsumedCapacityUnit{}}

	if request.UpdateRowChange.ReturnType == ReturnType_RT_AFTER_MODIFY {
		content := otsprotocol.ReturnContent{ReturnType: otsprotocol.ReturnType_RT_AFTER_MODIFY.Enum()}
		for _, column := range request.UpdateRowChange.ColumnNamesToReturn {
			content.ReturnColumnNames = append(content.ReturnColumnNames, column)
		}
		req.ReturnContent = &content
	}

	if err := tableStoreClient.doRequestWithRetry(updateRowUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	if request.UpdateRowChange.ReturnType == ReturnType_RT_AFTER_MODIFY {
		plainbufferRow, err := readRowsWithHeader(bytes.NewReader(resp.Row))
		if err != nil {
			return response, err
		}
		for _, cell := range plainbufferRow[0].cells {
			attribute := &AttributeColumn{ColumnName: string(cell.cellName), Value: cell.cellValue.Value, Timestamp: cell.cellTimestamp}
			response.Columns = append(response.Columns, attribute)
		}
	}

	response.ConsumedCapacityUnit.Read = *resp.Consumed.CapacityUnit.Read
	response.ConsumedCapacityUnit.Write = *resp.Consumed.CapacityUnit.Write
	return response, nil
}

// Batch Get Row
// @param BatchGetRowRequest
func (tableStoreClient *TableStoreClient) BatchGetRow(request *BatchGetRowRequest) (*BatchGetRowResponse, error) {
	req := new(otsprotocol.BatchGetRowRequest)

	var tablesInBatch []*otsprotocol.TableInBatchGetRowRequest

	for _, Criteria := range request.MultiRowQueryCriteria {
		table := new(otsprotocol.TableInBatchGetRowRequest)
		table.TableName = proto.String(Criteria.TableName)
		table.ColumnsToGet = Criteria.ColumnsToGet

		if Criteria.StartColumn != nil {
			table.StartColumn = Criteria.StartColumn
		}

		if Criteria.EndColumn != nil {
			table.EndColumn = Criteria.EndColumn
		}

		if Criteria.Filter != nil {
			table.Filter = Criteria.Filter.Serialize()
		}

		if Criteria.MaxVersion != 0 {
			table.MaxVersions = proto.Int32(int32(Criteria.MaxVersion))
		}

		if Criteria.TimeRange != nil {
			if Criteria.TimeRange.Specific != 0 {
				table.TimeRange = &otsprotocol.TimeRange{SpecificTime: proto.Int64(Criteria.TimeRange.Specific)}
			} else {
				table.TimeRange = &otsprotocol.TimeRange{StartTime: proto.Int64(Criteria.TimeRange.Start), EndTime: proto.Int64(Criteria.TimeRange.End)}
			}
		} else if Criteria.MaxVersion == 0 {
			return nil, errInvalidInput
		}

		for _, pk := range Criteria.PrimaryKey {
			pkWithBytes := pk.Build(false)
			table.PrimaryKey = append(table.PrimaryKey, pkWithBytes)
		}

		tablesInBatch = append(tablesInBatch, table)
	}

	req.Tables = tablesInBatch
	resp := new(otsprotocol.BatchGetRowResponse)

	response := &BatchGetRowResponse{TableToRowsResult: make(map[string][]RowResult)}
	if err := tableStoreClient.doBatchRequestWithRetry(batchGetRowUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	for _, table := range resp.Tables {
		index := int32(0)
		for _, row := range table.Rows {
			rowResult := &RowResult{TableName: *table.TableName, IsSucceed: *row.IsOk, ConsumedCapacityUnit: &ConsumedCapacityUnit{}, Index: index}
			index++
			if *row.IsOk == false {
				rowResult.Error = Error{Code: *row.Error.Code, Message: *row.Error.Message}
			} else {
				// len == 0 means row not exist
				if len(row.Row) > 0 {
					rows, err := readRowsWithHeader(bytes.NewReader(row.Row))
					if err != nil {
						return nil, err
					}

					for _, pk := range rows[0].primaryKey {
						pkColumn := &PrimaryKeyColumn{ColumnName: string(pk.cellName), Value: pk.cellValue.Value}
						rowResult.PrimaryKey.PrimaryKeys = append(rowResult.PrimaryKey.PrimaryKeys, pkColumn)
					}

					for _, cell := range rows[0].cells {
						dataColumn := &AttributeColumn{ColumnName: string(cell.cellName), Value: cell.cellValue.Value, Timestamp: cell.cellTimestamp}
						rowResult.Columns = append(rowResult.Columns, dataColumn)
					}
				}

				rowResult.ConsumedCapacityUnit.Read = *row.Consumed.CapacityUnit.Read
				rowResult.ConsumedCapacityUnit.Write = *row.Consumed.CapacityUnit.Write
			}

			response.TableToRowsResult[*table.TableName] = append(response.TableToRowsResult[*table.TableName], *rowResult)
		}

	}
	return response, nil
}

// Batch Write Row
// @param BatchWriteRowRequest
func (tableStoreClient *TableStoreClient) BatchWriteRow(request *BatchWriteRowRequest) (*BatchWriteRowResponse, error) {
	req := new(otsprotocol.BatchWriteRowRequest)

	var tablesInBatch []*otsprotocol.TableInBatchWriteRowRequest

	for key, value := range request.RowChangesGroupByTable {
		table := new(otsprotocol.TableInBatchWriteRowRequest)
		table.TableName = proto.String(key)

		for _, row := range value {
			rowInBatch := &otsprotocol.RowInBatchWriteRowRequest{}
			rowInBatch.Condition = row.getCondition()
			rowInBatch.RowChange = row.Serialize()
			rowInBatch.Type = row.getOperationType().Enum()

			if *rowInBatch.Type != otsprotocol.OperationType_DELETE {
				returnType := ReturnType_RT_NONE
				switch change := row.(type) {
				case *PutRowChange:
					returnType = change.ReturnType
				case *UpdateRowChange:
					returnType = change.ReturnType
				}
				switch returnType {
				case ReturnType_RT_PK:
					rowInBatch.ReturnContent = &otsprotocol.ReturnContent{
						ReturnType: otsprotocol.ReturnType_RT_PK.Enum(),
					}
				case ReturnType_RT_AFTER_MODIFY:
					updateRow, isUpdateRow := row.(*UpdateRowChange)
					if isUpdateRow {
						content := otsprotocol.ReturnContent{ReturnType: otsprotocol.ReturnType_RT_AFTER_MODIFY.Enum()}
						for _, column := range updateRow.ColumnNamesToReturn {
							content.ReturnColumnNames = append(content.ReturnColumnNames, column)
						}
						rowInBatch.ReturnContent = &content
					}
				}

			}
			table.Rows = append(table.Rows, rowInBatch)
		}

		tablesInBatch = append(tablesInBatch, table)
	}

	req.Tables = tablesInBatch
	req.IsAtomic = proto.Bool(request.IsAtomic)

	resp := new(otsprotocol.BatchWriteRowResponse)
	response := &BatchWriteRowResponse{TableToRowsResult: make(map[string][]RowResult)}

	if err := tableStoreClient.doBatchRequestWithRetry(batchWriteRowUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	for tableIndex, table := range resp.Tables {
		for index, row := range table.Rows {
			rowResult := &RowResult{TableName: *table.TableName, IsSucceed: *row.IsOk, ConsumedCapacityUnit: &ConsumedCapacityUnit{}, Index: int32(index)}
			if *row.IsOk == false {
				rowResult.Error = Error{Code: *row.Error.Code, Message: *row.Error.Message}
			} else {
				content := req.Tables[tableIndex].Rows[index].ReturnContent
				if content != nil {
					rows, err := readRowsWithHeader(bytes.NewReader(row.Row))
					if err != nil {
						return nil, err
					}

					if *content.ReturnType == otsprotocol.ReturnType_RT_PK {
						for _, pk := range rows[0].primaryKey {
							pkColumn := &PrimaryKeyColumn{ColumnName: string(pk.cellName), Value: pk.cellValue.Value}
							rowResult.PrimaryKey.PrimaryKeys = append(rowResult.PrimaryKey.PrimaryKeys, pkColumn)
						}
					}

					if *content.ReturnType == otsprotocol.ReturnType_RT_AFTER_MODIFY {
						for _, cell := range rows[0].cells {
							dataColumn := &AttributeColumn{ColumnName: string(cell.cellName), Value: cell.cellValue.Value, Timestamp: cell.cellTimestamp}
							rowResult.Columns = append(rowResult.Columns, dataColumn)
						}
					}
				}
				rowResult.ConsumedCapacityUnit.Read = *row.Consumed.CapacityUnit.Read
				rowResult.ConsumedCapacityUnit.Write = *row.Consumed.CapacityUnit.Write
			} /*else {
				rows, err := readRowsWithHeader(bytes.NewReader(row.Row))
				if err != nil {
					return nil, err
				}

				for _, pk := range (rows[0].primaryKey) {
					pkColumn := &PrimaryKeyColumn{ColumnName: string(pk.cellName), Value: pk.cellValue.Value}
					rowResult.PrimaryKey.PrimaryKeys = append(rowResult.PrimaryKey.PrimaryKeys, pkColumn)
				}

				for _, cell := range (rows[0].cells) {
					dataColumn := &DataColumn{ColumnName: string(cell.cellName), Value: cell.cellValue.Value}
					rowResult.Columns = append(rowResult.Columns, dataColumn)
				}

				rowResult.ConsumedCapacityUnit.Read = *row.Consumed.CapacityUnit.Read
				rowResult.ConsumedCapacityUnit.Write = *row.Consumed.CapacityUnit.Write
			}*/

			response.TableToRowsResult[*table.TableName] = append(response.TableToRowsResult[*table.TableName], *rowResult)
		}
	}
	return response, nil
}

// Get Range
// @param GetRangeRequest
func (tableStoreClient *TableStoreClient) GetRange(request *GetRangeRequest) (*GetRangeResponse, error) {
	req := new(otsprotocol.GetRangeRequest)
	req.TableName = proto.String(request.RangeRowQueryCriteria.TableName)
	req.Direction = request.RangeRowQueryCriteria.Direction.ToDirection().Enum()

	if request.RangeRowQueryCriteria.MaxVersion != 0 {
		req.MaxVersions = proto.Int32(request.RangeRowQueryCriteria.MaxVersion)
	}

	if request.RangeRowQueryCriteria.TransactionId != nil {
		req.TransactionId = request.RangeRowQueryCriteria.TransactionId
	}

	if request.RangeRowQueryCriteria.TimeRange != nil {
		if request.RangeRowQueryCriteria.TimeRange.Specific != 0 {
			req.TimeRange = &otsprotocol.TimeRange{SpecificTime: proto.Int64(request.RangeRowQueryCriteria.TimeRange.Specific)}
		} else {
			req.TimeRange = &otsprotocol.TimeRange{StartTime: proto.Int64(request.RangeRowQueryCriteria.TimeRange.Start), EndTime: proto.Int64(request.RangeRowQueryCriteria.TimeRange.End)}
		}
	} else if request.RangeRowQueryCriteria.MaxVersion == 0 {
		return nil, errInvalidInput
	}

	if request.RangeRowQueryCriteria.Limit != 0 {
		req.Limit = proto.Int32(request.RangeRowQueryCriteria.Limit)
	}

	if (request.RangeRowQueryCriteria.ColumnsToGet != nil) && len(request.RangeRowQueryCriteria.ColumnsToGet) > 0 {
		req.ColumnsToGet = request.RangeRowQueryCriteria.ColumnsToGet
	}

	if request.RangeRowQueryCriteria.Filter != nil {
		req.Filter = request.RangeRowQueryCriteria.Filter.Serialize()
	}

	if request.RangeRowQueryCriteria.StartColumn != nil {
		req.StartColumn = request.RangeRowQueryCriteria.StartColumn
	}

	if request.RangeRowQueryCriteria.EndColumn != nil {
		req.EndColumn = request.RangeRowQueryCriteria.EndColumn
	}

	req.DataBlockTypeHint = toPBDataBlockType(request.RangeRowQueryCriteria.DataBlockType)
	req.CompressTypeHint = toPBCompressType(request.RangeRowQueryCriteria.CompressType)
	if request.RangeRowQueryCriteria.ReturnSpecifiedPkOnly {
		req.ReturnEntirePrimaryKeys = proto.Bool(false)
	}

	req.InclusiveStartPrimaryKey = request.RangeRowQueryCriteria.StartPrimaryKey.Build(false)
	req.ExclusiveEndPrimaryKey = request.RangeRowQueryCriteria.EndPrimaryKey.Build(false)

	resp := new(otsprotocol.GetRangeResponse)
	response := &GetRangeResponse{ConsumedCapacityUnit: &ConsumedCapacityUnit{}}
	if err := tableStoreClient.doRequestWithRetry(getRangeUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.ConsumedCapacityUnit.Read = *resp.Consumed.CapacityUnit.Read
	response.ConsumedCapacityUnit.Write = *resp.Consumed.CapacityUnit.Write

	compressType, err := parseProtocolCompressType(resp.GetCompressType())
	if err != nil {
		return nil, err
	}
	response.CompressType = compressType

	if len(resp.NextStartPrimaryKey) != 0 {
		currentRows, err := readRowsWithHeader(bytes.NewReader(resp.NextStartPrimaryKey))
		if err != nil {
			return nil, err
		}

		response.NextStartPrimaryKey = &PrimaryKey{}
		for _, pk := range currentRows[0].primaryKey {
			pkColumn := &PrimaryKeyColumn{ColumnName: string(pk.cellName), Value: pk.cellValue.Value}
			response.NextStartPrimaryKey.PrimaryKeys = append(response.NextStartPrimaryKey.PrimaryKeys, pkColumn)
		}
	}

	if len(resp.Rows) == 0 {
		return response, nil
	}

	switch resp.GetDataBlockType() {
	case otsprotocol.DataBlockType_DBT_PLAIN_BUFFER:
		rows, err := parsePlainBufferRows(resp.Rows)
		if err != nil {
			return nil, err
		}
		response.Rows = rows
		response.DataBlockType = PlainBuffer
	case otsprotocol.DataBlockType_DBT_SIMPLE_ROW_MATRIX:
		rows, err := parseMatrixRows(resp.Rows)
		if err != nil {
			return nil, err
		}
		response.Rows = rows
		response.DataBlockType = SimpleRowMatrix
	default:
		return nil, fmt.Errorf("unknow data block type %d", resp.GetDataBlockType())
	}
	return response, nil
}

func toPBCompressType(compressType CompressType) *otsprotocol.CompressType {
	switch compressType {
	case None:
		return otsprotocol.CompressType_CPT_NONE.Enum()
	default:
		// return CompressType_CPT_NONE
		return otsprotocol.CompressType_CPT_NONE.Enum()
	}
}

func toPBDataBlockType(blockType DataBlockType) *otsprotocol.DataBlockType {
	switch blockType {
	case PlainBuffer:
		return otsprotocol.DataBlockType_DBT_PLAIN_BUFFER.Enum()
	case SimpleRowMatrix:
		return otsprotocol.DataBlockType_DBT_SIMPLE_ROW_MATRIX.Enum()
	default:
		// return DataBlockType_DBT_PLAIN_BUFFER
		return otsprotocol.DataBlockType_DBT_PLAIN_BUFFER.Enum()
	}
}

func parsePlainBufferRows(rowBytes []byte) ([]*Row, error) {
	pbRows, err := readRowsWithHeader(bytes.NewReader(rowBytes))
	if err != nil {
		return nil, err
	}

	rows := make([]*Row, len(pbRows))
	for i, row := range pbRows {
		currentRow := new(Row)
		currentpk := new(PrimaryKey)
		for _, pk := range row.primaryKey {
			pkColumn := &PrimaryKeyColumn{ColumnName: string(pk.cellName), Value: pk.cellValue.Value}
			currentpk.PrimaryKeys = append(currentpk.PrimaryKeys, pkColumn)
		}

		currentRow.PrimaryKey = currentpk

		for _, cell := range row.cells {
			dataColumn := &AttributeColumn{ColumnName: string(cell.cellName), Value: cell.cellValue.Value, Timestamp: cell.cellTimestamp}
			currentRow.Columns = append(currentRow.Columns, dataColumn)
		}

		rows[i] = currentRow
	}
	return rows, nil
}

func (client *TableStoreClient) SQLQuery(req *SQLQueryRequest) (*SQLQueryResponse, error) {
	// create request
	pbReq := &otsprotocol.SQLQueryRequest{}
	pbReq.Query = &req.Query
	pbReq.Version = otsprotocol.SQLPayloadVersion_SQL_FLAT_BUFFERS.Enum()
	pbReq.SqlVersion = proto.Int64(1)

	// do request
	pbResp := otsprotocol.SQLQueryResponse{}
	response := &SQLQueryResponse{SQLQueryConsumed: &SQLQueryConsumed{}}
	if err := client.doRequestWithRetry(sqlQueryUri, pbReq, &pbResp, &response.ResponseInfo, req.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.StmtType = formatSQLStmtTypeFromPB(pbResp.GetType())
	if pbResp.GetVersion() == otsprotocol.SQLPayloadVersion_SQL_PLAIN_BUFFER {
		rs, err := NewSQLResultSetFromPlainBuffer(pbResp.Rows)
		if err != nil {
			return nil, err
		}
		response.ResultSet = rs
		response.PayloadVersion = SQLPAYLOAD_PLAIN_BUFFER
	} else if pbResp.GetVersion() == otsprotocol.SQLPayloadVersion_SQL_FLAT_BUFFERS {
		rs, err := NewSQLResultSetFromFlatBuffers(pbResp.Rows)
		if err != nil {
			return nil, err
		}
		response.ResultSet = rs
		response.PayloadVersion = SQLPAYLOAD_FLAT_BUFFERS
	}

	tableConsumes := make([]*TableConsumedCU, 0, len(pbResp.Consumes))
	for _, consume := range pbResp.Consumes {
		tableConsume := new(TableConsumedCU)
		tableConsume.TableName = consume.GetTableName()
		if consume.Consumed != nil && consume.Consumed.CapacityUnit != nil {
			tableConsume.ConsumedCapacityUnit = new(ConsumedCapacityUnit)
			tableConsume.ConsumedCapacityUnit.Read = consume.Consumed.CapacityUnit.GetRead()
			tableConsume.ConsumedCapacityUnit.Write = consume.Consumed.CapacityUnit.GetWrite()
		}

		tableConsumes = append(tableConsumes, tableConsume)
	}
	response.SQLQueryConsumed.TableConsumes = tableConsumes

	searchConsumes := make([]*SearchConsumedCU, 0, len(pbResp.SearchConsumes))
	for _, consume := range pbResp.SearchConsumes {
		searchConsume := new(SearchConsumedCU)
		searchConsume.TableName = consume.GetTableName()
		searchConsume.IndexName = consume.GetIndexName()
		if consume.Consumed != nil && consume.Consumed.CapacityUnit != nil {
			searchConsume.ConsumedCapacityUnit = new(ConsumedCapacityUnit)
			searchConsume.ConsumedCapacityUnit.Read = consume.Consumed.CapacityUnit.GetRead()
			searchConsume.ConsumedCapacityUnit.Write = consume.Consumed.CapacityUnit.GetWrite()
		}

		searchConsumes = append(searchConsumes, searchConsume)
	}
	response.SQLQueryConsumed.SearchConsumes = searchConsumes

	return response, nil
}

func (client *TableStoreClient) ListStream(req *ListStreamRequest) (*ListStreamResponse, error) {
	pbReq := &otsprotocol.ListStreamRequest{}
	pbReq.TableName = req.TableName

	pbResp := otsprotocol.ListStreamResponse{}
	resp := ListStreamResponse{}
	if err := client.doRequestWithRetry(listStreamUri, pbReq, &pbResp, &resp.ResponseInfo, req.ExtraRequestInfo); err != nil {
		return nil, err
	}

	streams := make([]Stream, len(pbResp.Streams))
	for i, pbStream := range pbResp.Streams {
		streams[i] = Stream{
			Id:           (*StreamId)(pbStream.StreamId),
			TableName:    pbStream.TableName,
			CreationTime: *pbStream.CreationTime}
	}
	resp.Streams = streams[:]
	return &resp, nil
}

func (client *TableStoreClient) DescribeStream(req *DescribeStreamRequest) (*DescribeStreamResponse, error) {
	pbReq := &otsprotocol.DescribeStreamRequest{}
	{
		pbReq.StreamId = (*string)(req.StreamId)
		pbReq.InclusiveStartShardId = (*string)(req.InclusiveStartShardId)
		pbReq.ShardLimit = req.ShardLimit
	}
	pbResp := otsprotocol.DescribeStreamResponse{}
	resp := DescribeStreamResponse{}
	if err := client.doRequestWithRetry(describeStreamUri, pbReq, &pbResp, &resp.ResponseInfo, req.ExtraRequestInfo); err != nil {
		return nil, err
	}

	resp.StreamId = (*StreamId)(pbResp.StreamId)
	resp.ExpirationTime = *pbResp.ExpirationTime
	resp.TableName = pbResp.TableName
	resp.CreationTime = *pbResp.CreationTime
	Assert(pbResp.StreamStatus != nil, "StreamStatus in DescribeStreamResponse is required.")
	switch *pbResp.StreamStatus {
	case otsprotocol.StreamStatus_STREAM_ENABLING:
		resp.Status = SS_Enabling
	case otsprotocol.StreamStatus_STREAM_ACTIVE:
		resp.Status = SS_Active
	}
	resp.NextShardId = (*ShardId)(pbResp.NextShardId)
	shards := make([]*StreamShard, len(pbResp.Shards))
	for i, pbShard := range pbResp.Shards {
		shards[i] = &StreamShard{
			SelfShard:   (*ShardId)(pbShard.ShardId),
			FatherShard: (*ShardId)(pbShard.ParentId),
			MotherShard: (*ShardId)(pbShard.ParentSiblingId)}
	}
	resp.Shards = shards[:]
	return &resp, nil
}

func (client *TableStoreClient) GetShardIterator(req *GetShardIteratorRequest) (*GetShardIteratorResponse, error) {
	pbReq := &otsprotocol.GetShardIteratorRequest{
		StreamId: (*string)(req.StreamId),
		ShardId:  (*string)(req.ShardId)}

	if req.Timestamp != nil {
		pbReq.Timestamp = req.Timestamp
	}

	if req.Token != nil {
		pbReq.Token = req.Token
	}

	pbResp := otsprotocol.GetShardIteratorResponse{}
	resp := GetShardIteratorResponse{}
	if err := client.doRequestWithRetry(getShardIteratorUri, pbReq, &pbResp, &resp.ResponseInfo, req.ExtraRequestInfo); err != nil {
		return nil, err
	}

	resp.ShardIterator = (*ShardIterator)(pbResp.ShardIterator)
	resp.Token = pbResp.NextToken
	return &resp, nil
}

func (client TableStoreClient) GetStreamRecord(req *GetStreamRecordRequest) (*GetStreamRecordResponse, error) {
	pbReq := &otsprotocol.GetStreamRecordRequest{
		ShardIterator: (*string)(req.ShardIterator)}
	if req.Limit != nil {
		pbReq.Limit = req.Limit
	}
	if req.TableName != nil {
		pbReq.TableName = req.TableName
	}

	pbResp := otsprotocol.GetStreamRecordResponse{}
	resp := GetStreamRecordResponse{}
	if err := client.doRequestWithRetry(getStreamRecordUri, pbReq, &pbResp, &resp.ResponseInfo, req.ExtraRequestInfo); err != nil {
		return nil, err
	}

	resp.MayMoreRecord = pbResp.MayMoreRecord
	if pbResp.GetConsumed() != nil && pbResp.GetConsumed().GetCapacityUnit() != nil {
		cu := pbResp.GetConsumed().GetCapacityUnit()
		resp.CapacityUnit = &ConsumedCapacityUnit{
			Read:  cu.GetRead(),
			Write: cu.GetWrite(),
		}
	}

	if pbResp.NextShardIterator != nil {
		resp.NextShardIterator = (*ShardIterator)(pbResp.NextShardIterator)
	}
	records := make([]*StreamRecord, len(pbResp.StreamRecords))
	for i, pbRecord := range pbResp.StreamRecords {
		record := StreamRecord{}
		records[i] = &record

		switch *pbRecord.ActionType {
		case otsprotocol.ActionType_PUT_ROW:
			record.Type = AT_Put
		case otsprotocol.ActionType_UPDATE_ROW:
			record.Type = AT_Update
		case otsprotocol.ActionType_DELETE_ROW:
			record.Type = AT_Delete
		}

		plainRows, err := readRowsWithHeader(bytes.NewReader(pbRecord.Record))
		if err != nil {
			return nil, err
		}
		Assert(len(plainRows) == 1,
			"There must be exactly one row in a StreamRecord.")
		plainRow := plainRows[0]
		pkey := PrimaryKey{}
		record.PrimaryKey = &pkey
		pkey.PrimaryKeys = make([]*PrimaryKeyColumn, len(plainRow.primaryKey))
		for i, pk := range plainRow.primaryKey {
			pkc := PrimaryKeyColumn{
				ColumnName: string(pk.cellName),
				Value:      pk.cellValue.Value}
			pkey.PrimaryKeys[i] = &pkc
		}
		Assert(plainRow.extension != nil,
			"extension in a stream record is required.")
		record.Info = plainRow.extension
		record.Columns = make([]*RecordColumn, len(plainRow.cells))
		for i, plainCell := range plainRow.cells {
			cell := RecordColumn{}
			record.Columns[i] = &cell

			name := string(plainCell.cellName)
			cell.Name = &name
			if plainCell.cellValue != nil {
				cell.Type = RCT_Put
			} else {
				if plainCell.cellTimestamp > 0 {
					cell.Type = RCT_DeleteOneVersion
				} else {
					cell.Type = RCT_DeleteAllVersions
				}
			}
			switch cell.Type {
			case RCT_Put:
				cell.Value = plainCell.cellValue.Value
				fallthrough
			case RCT_DeleteOneVersion:
				cell.Timestamp = &plainCell.cellTimestamp
			case RCT_DeleteAllVersions:
				break
			}
		}

		if pbRecord.GetOriginRecord() != nil {
			originPlainRows, err := readRowsWithHeader(bytes.NewReader(pbRecord.GetOriginRecord()))
			if err != nil {
				return nil, err
			}
			Assert(len(originPlainRows) == 1,
				"There must be exactly one row in a StreamRecord.")
			originPlainRow := originPlainRows[0]

			record.OriginColumns = make([]*RecordColumn, len(originPlainRow.cells))
			for i, plainCell := range originPlainRow.cells {
				cell := RecordColumn{}
				record.OriginColumns[i] = &cell

				name := string(plainCell.cellName)
				cell.Name = &name
				if plainCell.cellValue != nil {
					cell.Type = RCT_Put
				} else {
					if plainCell.cellTimestamp > 0 {
						cell.Type = RCT_DeleteOneVersion
					} else {
						cell.Type = RCT_DeleteAllVersions
					}
				}
				switch cell.Type {
				case RCT_Put:
					cell.Value = plainCell.cellValue.Value
					fallthrough
				case RCT_DeleteOneVersion:
					cell.Timestamp = &plainCell.cellTimestamp
				case RCT_DeleteAllVersions:
					break
				}
			}
		}

	}
	resp.Records = records
	return &resp, nil
}

func (client TableStoreClient) ComputeSplitPointsBySize(req *ComputeSplitPointsBySizeRequest) (*ComputeSplitPointsBySizeResponse, error) {
	pbReq := &otsprotocol.ComputeSplitPointsBySizeRequest{
		TableName:           &(req.TableName),
		SplitSize:           &(req.SplitSize),
		SplitSizeUnitInByte: req.SplitSizeUnitInByte,
		SplitPointLimit:     req.SplitPointLimit,
	}

	pbResp := otsprotocol.ComputeSplitPointsBySizeResponse{}
	resp := ComputeSplitPointsBySizeResponse{}
	if err := client.doRequestWithRetry(computeSplitPointsBySizeRequestUri, pbReq, &pbResp, &resp.ResponseInfo, req.ExtraRequestInfo); err != nil {
		return nil, err
	}

	beginPk := &PrimaryKey{}
	endPk := &PrimaryKey{}
	for _, pkSchema := range pbResp.Schema {
		beginPk.AddPrimaryKeyColumnWithMinValue(*pkSchema.Name)
		endPk.AddPrimaryKeyColumnWithMaxValue(*pkSchema.Name)
	}
	lastPk := beginPk
	nowPk := endPk

	for _, pbRecord := range pbResp.SplitPoints {
		plainRows, err := readRowsWithHeader(bytes.NewReader(pbRecord))
		if err != nil {
			return nil, err
		}

		nowPk = &PrimaryKey{}
		for _, pk := range plainRows[0].primaryKey {
			nowPk.AddPrimaryKeyColumn(string(pk.cellName), pk.cellValue.Value)
		}

		for i := len(plainRows[0].primaryKey); i < len(pbResp.GetSchema()); i++ {
			nowPk.AddPrimaryKeyColumnWithMinValue(*pbResp.Schema[i].Name)
		}

		newSplit := &Split{LowerBound: lastPk, UpperBound: nowPk}
		resp.Splits = append(resp.Splits, newSplit)
		lastPk = nowPk

	}

	newSplit := &Split{LowerBound: lastPk, UpperBound: endPk}
	resp.Splits = append(resp.Splits, newSplit)

	index := 0
	for _, pbLocation := range pbResp.Locations {
		count := *pbLocation.Repeat
		value := *pbLocation.Location

		for i := int64(0); i < count; i++ {
			resp.Splits[index].Location = value
			index++
		}
	}
	return &resp, nil
}

func (client *TableStoreClient) StartLocalTransaction(request *StartLocalTransactionRequest) (*StartLocalTransactionResponse, error) {
	req := new(otsprotocol.StartLocalTransactionRequest)
	resp := new(otsprotocol.StartLocalTransactionResponse)

	req.TableName = proto.String(request.TableName)
	req.Key = request.PrimaryKey.Build(false)

	response := &StartLocalTransactionResponse{}
	if err := client.doRequestWithRetry(createlocaltransactionuri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	response.TransactionId = resp.TransactionId
	return response, nil
}

func (client *TableStoreClient) CommitTransaction(request *CommitTransactionRequest) (*CommitTransactionResponse, error) {
	req := new(otsprotocol.CommitTransactionRequest)
	resp := new(otsprotocol.CommitTransactionResponse)

	req.TransactionId = request.TransactionId

	response := &CommitTransactionResponse{}
	if err := client.doRequestWithRetry(committransactionuri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

func (client *TableStoreClient) AbortTransaction(request *AbortTransactionRequest) (*AbortTransactionResponse, error) {
	req := new(otsprotocol.AbortTransactionRequest)
	resp := new(otsprotocol.AbortTransactionResponse)

	req.TransactionId = request.TransactionId

	response := &AbortTransactionResponse{}
	if err := client.doRequestWithRetry(aborttransactionuri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	return response, nil
}

func (client *TableStoreClient) CreateDeliveryTask(request *CreateDeliveryTaskRequest) (*CreateDeliveryTaskResponse, error) {
	pbReq := &otsprotocol.CreateDeliveryTaskRequest{
		TableName:  &request.TableName,
		TaskName:   &request.TaskName,
		TaskType:   otsprotocol.DeliveryTaskType(request.TaskType).Enum(),
		TaskConfig: toTaskPbConfig(request.TaskConfig),
	}
	pbResp := new(otsprotocol.CreateDeliveryTaskResponse)
	response := new(CreateDeliveryTaskResponse)
	if err := client.doRequestWithRetry(createDeliveryTaskUri, pbReq, pbResp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	return response, nil
}

func (client *TableStoreClient) DeleteDeliveryTask(request *DeleteDeliveryTaskRequest) (*DeleteDeliveryTaskResponse, error) {
	pbReq := &otsprotocol.DeleteDeliveryTaskRequest{
		TableName: &request.TableName,
		TaskName:  &request.TaskName,
	}
	pbResp := new(otsprotocol.DeleteDeliveryTaskResponse)
	response := new(DeleteDeliveryTaskResponse)
	if err := client.doRequestWithRetry(deleteDeliveryTaskUri, pbReq, pbResp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	return response, nil
}

func (client *TableStoreClient) ListDeliveryTask(request *ListDeliveryTaskRequest) (*ListDeliveryTaskResponse, error) {
	pbReq := &otsprotocol.ListDeliveryTaskRequest{
		TableName: &request.TableName,
	}
	pbResp := new(otsprotocol.ListDeliveryTaskResponse)
	response := new(ListDeliveryTaskResponse)
	if err := client.doRequestWithRetry(listDeliveryTaskUri, pbReq, pbResp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	response.Tasks = make([]*DeliveryTaskInfo, len(pbResp.Tasks))
	for i, task := range pbResp.Tasks {
		response.Tasks[i] = &DeliveryTaskInfo{
			TableName: task.GetTableName(),
			TaskName:  task.GetTaskName(),
			TaskType:  TaskType(task.GetTaskType()),
		}
	}
	return response, nil
}

func (client *TableStoreClient) DescribeDeliveryTask(request *DescribeDeliveryTaskRequest) (*DescribeDeliveryTaskResponse, error) {
	pbReq := &otsprotocol.DescribeDeliveryTaskRequest{
		TableName: &request.TableName,
		TaskName:  &request.TaskName,
	}
	pbResp := new(otsprotocol.DescribeDeliveryTaskResponse)
	response := new(DescribeDeliveryTaskResponse)
	if err := client.doRequestWithRetry(describeDeliveryTaskUri, pbReq, pbResp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}
	response.TaskType = TaskType(pbResp.GetTaskType())
	response.TaskConfig = toOSSTaskConfig(pbResp.TaskConfig)
	response.TaskSyncStat = toTaskSyncStat(pbResp.TaskSyncStat)
	return response, nil
}

func (client *TableStoreClient) ComputeSplits(request *ComputeSplitsRequest) (*ComputeSplitsResponse, error) {
	req := new(otsprotocol.ComputeSplitsRequest)
	resp := new(otsprotocol.ComputeSplitsResponse)

	req.TableName = proto.String(request.TableName)
	req.SearchIndexSplitsOptions = new(otsprotocol.SearchIndexSplitsOptions)
	if request.searchIndexSplitsOptions != nil {
		req.SearchIndexSplitsOptions.IndexName = proto.String(request.searchIndexSplitsOptions.IndexName)
	}

	response := &ComputeSplitsResponse{}
	if err := client.doRequestWithRetry(computeSplitsUri, req, resp, &response.ResponseInfo, request.ExtraRequestInfo); err != nil {
		return nil, err
	}

	if resp.SessionId != nil && len(resp.SessionId) > 0 {
		response.SessionId = resp.SessionId
	}
	if resp.SplitsSize != nil {
		response.SplitsSize = *resp.SplitsSize
	}

	return response, nil
}