dps-sdk/dpsv1/client.go

package dpsv1

import (
	"context"
	"encoding/json"
	"errors"
	"git.sxidc.com/service-supports/dps-sdk/client"
	"git.sxidc.com/service-supports/dps-sdk/pb/v1"
	"git.sxidc.com/service-supports/dps-sdk/pb/v1/request"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"io"
	"sync"
)

type Client struct {
	databaseID              string
	conn                    *grpc.ClientConn
	commandServiceClient    v1.CommandServiceClient
	queryServiceClient      v1.QueryServiceClient
	eventQueryServiceClient v1.EventQueryServiceClient
	transactionMutex        sync.Mutex
}

func NewClient(address string, databaseID string) (*Client, error) {
	conn, err := grpc.DialContext(context.Background(), address,
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		return nil, err
	}

	return &Client{
		databaseID:              databaseID,
		conn:                    conn,
		commandServiceClient:    v1.NewCommandServiceClient(conn),
		queryServiceClient:      v1.NewQueryServiceClient(conn),
		eventQueryServiceClient: v1.NewEventQueryServiceClient(conn),
	}, nil
}

func DestroyClient(client *Client) error {
	if client == nil {
		return nil
	}

	client.transactionMutex.Lock()
	defer client.transactionMutex.Unlock()

	err := client.conn.Close()
	if err != nil {
		return err
	}

	client.databaseID = ""
	client.conn = nil
	client.commandServiceClient = nil
	client.queryServiceClient = nil

	return nil
}

func (c *Client) AutoMigrate(req *client.AutoMigrateRequest) error {
	items := make([]*request.AutoMigrateItem, 0)
	for _, reqItem := range req.Items {
		tableModelDescribeJsonBytes, err := json.Marshal(reqItem.TableModelDescribe)
		if err != nil {
			return err
		}

		items = append(items, &request.AutoMigrateItem{
			TablePrefixWithSchema: reqItem.TablePrefixWithSchema,
			Version:               reqItem.Version,
			TableModelDescribe:    tableModelDescribeJsonBytes,
		})
	}

	_, err := c.commandServiceClient.AutoMigrate(context.Background(), &request.AutoMigrateRequest{
		DatabaseID: c.databaseID,
		Items:      items,
	})
	if err != nil {
		return err
	}

	return nil
}

func (c *Client) Transaction(txFunc client.TransactionFunc) error {
	stream, err := c.commandServiceClient.Transaction(context.Background())
	if err != nil {
		return err
	}

	defer func() {
		innerErr := stream.CloseSend()
		if innerErr != nil {
			panic(innerErr)
		}
	}()

	err = stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_TransactionBeginRequest{
			TransactionBeginRequest: &request.TransactionBeginRequest{DatabaseID: c.databaseID},
		}})
	if err != nil {
		return err
	}

	err = txFunc(&Transaction{
		stream: stream,
		client: c,
	})
	if err != nil {
		return err
	}

	err = stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_TransactionEndRequest{
			TransactionEndRequest: &request.TransactionEndRequest{},
		}})
	if err != nil {
		return err
	}

	_, err = stream.Recv()
	if err != nil && err != io.EOF {
		return err
	}

	return nil
}

func (c *Client) Insert(req *client.InsertRequest) (string, error) {
	if req.TableRow == nil {
		return "", nil
	}

	tableRow, err := req.TableRow.ToDPSTableRow()
	if err != nil {
		return "", err
	}

	reply, err := c.commandServiceClient.Insert(context.Background(), &request.InsertRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		KeyColumns:            req.KeyColumns,
		TableRow:              tableRow,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) InsertBatch(req *client.InsertBatchRequest) (string, error) {
	tableRowItems := make([]*request.InsertTableRowItem, 0)

	for _, reqTableItem := range req.Items {
		tableRows := make([]*request.TableRow, 0)
		for _, reqTableRow := range reqTableItem.TableRows {
			if reqTableRow == nil {
				continue
			}

			tableRow, err := reqTableRow.ToDPSTableRow()
			if err != nil {
				return "", err
			}

			tableRows = append(tableRows, tableRow)
		}

		tableRowItems = append(tableRowItems, &request.InsertTableRowItem{
			TablePrefixWithSchema: reqTableItem.TablePrefixWithSchema,
			Version:               reqTableItem.Version,
			KeyColumns:            reqTableItem.KeyColumns,
			TableRows:             tableRows,
		})
	}

	reply, err := c.commandServiceClient.InsertBatch(context.Background(), &request.InsertBatchRequest{
		DatabaseID: c.databaseID,
		Items:      tableRowItems,
		UserID:     req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) Delete(req *client.DeleteRequest) (string, error) {
	reply, err := c.commandServiceClient.Delete(context.Background(), &request.DeleteRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		KeyValues:             req.KeyValues,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) DeleteWhere(req *client.DeleteWhereRequest) (string, error) {
	if req.Where == nil {
		return "", errors.New("没有传递Where条件")
	}

	whereJsonBytes, err := req.Where.ToJson()
	if err != nil {
		return "", err
	}

	reply, err := c.commandServiceClient.DeleteWhere(context.Background(), &request.DeleteWhereRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		KeyColumns:            req.KeyColumns,
		Where:                 whereJsonBytes,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) DeleteBatch(req *client.DeleteBatchRequest) (string, error) {
	tableRowItems := make([]*request.DeleteTableRowItem, 0)

	for _, reqTableItem := range req.Items {
		items := make([]*request.DeleteItem, 0)
		for _, reqKeyValues := range reqTableItem.KeyValues {
			items = append(items, &request.DeleteItem{
				KeyValues: reqKeyValues,
			})
		}

		tableRowItems = append(tableRowItems, &request.DeleteTableRowItem{
			TablePrefixWithSchema: reqTableItem.TablePrefixWithSchema,
			Version:               reqTableItem.Version,
			Items:                 items,
		})
	}

	reply, err := c.commandServiceClient.DeleteBatch(context.Background(), &request.DeleteBatchRequest{
		DatabaseID: c.databaseID,
		Items:      tableRowItems,
		UserID:     req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) Update(req *client.UpdateRequest) (string, error) {
	if req.NewTableRow == nil {
		return "", nil
	}

	newTableRow, err := req.NewTableRow.ToDPSTableRow()
	if err != nil {
		return "", err
	}

	reply, err := c.commandServiceClient.Update(context.Background(), &request.UpdateRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		KeyValues:             req.KeyValues,
		NewTableRow:           newTableRow,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) Replay(req *client.ReplayRequest) (string, error) {
	reply, err := c.commandServiceClient.Replay(context.Background(), &request.ReplayRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		KeyValues:             req.KeyValues,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) QueryByWhereAndOrderBy(req *client.QueryByWhereAndOrderByRequest) (string, []client.TableRow, int64, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		selectJsonBytes = innerJsonBytes
	}

	var whereJsonBytes []byte
	if req.Where != nil {
		innerJsonBytes, err := req.Where.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		whereJsonBytes = innerJsonBytes
	}

	reply, err := c.queryServiceClient.QueryByWhereAndOrderBy(context.Background(), &request.QueryByWhereAndOrderByRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Select:                selectJsonBytes,
		Where:                 whereJsonBytes,
		OrderBy:               req.OrderBy,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, 0, err
	}

	tableRows, err := client.FromDSPInfosData(reply.Infos)
	if err != nil {
		return "", nil, 0, err
	}

	return reply.Statement, tableRows, reply.TotalCount, nil
}

func (c *Client) CommonQuery(req *client.CommonQueryRequest) (string, []client.TableRow, int64, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		selectJsonBytes = innerJsonBytes
	}

	var whereJsonBytes []byte
	if req.Where != nil {
		innerJsonBytes, err := req.Where.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		whereJsonBytes = innerJsonBytes
	}

	var orJsonBytes []byte
	if req.Or != nil {
		innerJsonBytes, err := req.Or.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		orJsonBytes = innerJsonBytes
	}

	var joinsJsonBytes []byte
	if req.Joins != nil {
		innerJsonBytes, err := req.Joins.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		joinsJsonBytes = innerJsonBytes
	}

	var havingJsonBytes []byte
	if req.Having != nil {
		innerJsonBytes, err := req.Having.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		havingJsonBytes = innerJsonBytes
	}

	reply, err := c.queryServiceClient.CommonQuery(context.Background(), &request.CommonQueryRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Select:                selectJsonBytes,
		Where:                 whereJsonBytes,
		OrderBy:               req.OrderBy,
		Or:                    orJsonBytes,
		GroupBy:               req.GroupBy,
		Joins:                 joinsJsonBytes,
		Having:                havingJsonBytes,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, 0, err
	}

	tableRows, err := client.FromDSPInfosData(reply.Infos)
	if err != nil {
		return "", nil, 0, err
	}

	return reply.Statement, tableRows, reply.TotalCount, nil
}

func (c *Client) QueryOnlyByWhereAndOrderBy(req *client.QueryByWhereAndOrderByRequest) (string, []client.TableRow, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, err
		}

		selectJsonBytes = innerJsonBytes
	}

	var whereJsonBytes []byte
	if req.Where != nil {
		innerJsonBytes, err := req.Where.ToJson()
		if err != nil {
			return "", nil, err
		}

		whereJsonBytes = innerJsonBytes
	}

	reply, err := c.queryServiceClient.QueryOnlyByWhereAndOrderBy(context.Background(), &request.QueryByWhereAndOrderByRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Select:                selectJsonBytes,
		Where:                 whereJsonBytes,
		OrderBy:               req.OrderBy,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, err
	}

	tableRows, err := client.FromDSPInfosData(reply.Infos)
	if err != nil {
		return "", nil, err
	}

	return reply.Statement, tableRows, nil
}

func (c *Client) CommonQueryOnly(req *client.CommonQueryRequest) (string, []client.TableRow, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, err
		}

		selectJsonBytes = innerJsonBytes
	}

	var whereJsonBytes []byte
	if req.Where != nil {
		innerJsonBytes, err := req.Where.ToJson()
		if err != nil {
			return "", nil, err
		}

		whereJsonBytes = innerJsonBytes
	}

	var orJsonBytes []byte
	if req.Or != nil {
		innerJsonBytes, err := req.Or.ToJson()
		if err != nil {
			return "", nil, err
		}

		orJsonBytes = innerJsonBytes
	}

	var joinsJsonBytes []byte
	if req.Joins != nil {
		innerJsonBytes, err := req.Joins.ToJson()
		if err != nil {
			return "", nil, err
		}

		joinsJsonBytes = innerJsonBytes
	}

	var havingJsonBytes []byte
	if req.Having != nil {
		innerJsonBytes, err := req.Having.ToJson()
		if err != nil {
			return "", nil, err
		}

		havingJsonBytes = innerJsonBytes
	}

	reply, err := c.queryServiceClient.CommonQueryOnly(context.Background(), &request.CommonQueryRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Select:                selectJsonBytes,
		Where:                 whereJsonBytes,
		OrderBy:               req.OrderBy,
		Or:                    orJsonBytes,
		GroupBy:               req.GroupBy,
		Joins:                 joinsJsonBytes,
		Having:                havingJsonBytes,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, err
	}

	tableRows, err := client.FromDSPInfosData(reply.Infos)
	if err != nil {
		return "", nil, err
	}

	return reply.Statement, tableRows, nil
}

func (c *Client) QueryByKeys(req *client.QueryByKeysRequest) (string, *client.TableRow, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.queryServiceClient.QueryByKeys(context.Background(), &request.QueryByKeysRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
	})
	if err != nil {
		return "", nil, err
	}

	tableRow := client.NewTableRow()
	err = tableRow.FromDSPInfoData(reply.Info)
	if err != nil {
		return "", nil, err
	}

	return reply.Statement, tableRow, nil
}

func (c *Client) CountWhere(req *client.CountWhereRequest) (string, int64, error) {
	var whereJsonBytes []byte
	if req.Where != nil {
		innerJsonBytes, err := req.Where.ToJson()
		if err != nil {
			return "", 0, err
		}

		whereJsonBytes = innerJsonBytes
	}

	reply, err := c.queryServiceClient.CountWhere(context.Background(), &request.CountWhereRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Where:                 whereJsonBytes,
	})
	if err != nil {
		return "", 0, err
	}

	return reply.Statement, reply.Count, nil
}

func (c *Client) CommonCount(req *client.CommonCountRequest) (string, int64, error) {
	var whereJsonBytes []byte
	if req.Where != nil {
		innerJsonBytes, err := req.Where.ToJson()
		if err != nil {
			return "", 0, err
		}

		whereJsonBytes = innerJsonBytes
	}

	var orJsonBytes []byte
	if req.Or != nil {
		innerJsonBytes, err := req.Or.ToJson()
		if err != nil {
			return "", 0, err
		}

		orJsonBytes = innerJsonBytes
	}

	var joinsJsonBytes []byte
	if req.Joins != nil {
		innerJsonBytes, err := req.Joins.ToJson()
		if err != nil {
			return "", 0, err
		}

		joinsJsonBytes = innerJsonBytes
	}

	var havingJsonBytes []byte
	if req.Having != nil {
		innerJsonBytes, err := req.Having.ToJson()
		if err != nil {
			return "", 0, err
		}

		havingJsonBytes = innerJsonBytes
	}

	reply, err := c.queryServiceClient.CommonCount(context.Background(), &request.CommonCountRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Where:                 whereJsonBytes,
		Or:                    orJsonBytes,
		GroupBy:               req.GroupBy,
		Joins:                 joinsJsonBytes,
		Having:                havingJsonBytes,
	})
	if err != nil {
		return "", 0, err
	}

	return reply.Statement, reply.Count, nil
}

func (c *Client) EventQueryByKeys(req *client.EventQueryByKeysRequest) (string, []client.EventInfo, int64, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.EventQueryByKeys(context.Background(), &request.EventQueryByKeysRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, 0, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), reply.TotalCount, nil
}

func (c *Client) CommonEventQuery(req *client.CommonEventQueryRequest) (string, []client.EventInfo, int64, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.CommonEventQuery(context.Background(), &request.CommonEventQueryRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		Version:               req.Version,
		Operation:             req.Operation,
		CreatorID:             req.CreatorID,
		StartCreatedTime:      req.StartCreatedTime,
		EndCreatedTime:        req.EndCreatedTime,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, 0, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), reply.TotalCount, nil
}

func (c *Client) EventQueryOnlyByKeys(req *client.EventQueryByKeysRequest) (string, []client.EventInfo, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.EventQueryOnlyByKeys(context.Background(), &request.EventQueryByKeysRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), nil
}

func (c *Client) CommonEventQueryOnly(req *client.CommonEventQueryRequest) (string, []client.EventInfo, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.CommonEventQueryOnly(context.Background(), &request.CommonEventQueryRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		Version:               req.Version,
		Operation:             req.Operation,
		CreatorID:             req.CreatorID,
		StartCreatedTime:      req.StartCreatedTime,
		EndCreatedTime:        req.EndCreatedTime,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), nil
}

func (c *Client) CountEventByKeys(req *client.CountEventByKeysRequest) (string, int64, error) {
	reply, err := c.eventQueryServiceClient.CountEventByKeys(context.Background(), &request.CountEventByKeysRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		KeyValues:             req.KeyValues,
	})
	if err != nil {
		return "", 0, err
	}

	return reply.Statement, reply.Count, nil
}

func (c *Client) CommonCountEvent(req *client.CommonCountEventRequest) (string, int64, error) {
	reply, err := c.eventQueryServiceClient.CommonCountEvent(context.Background(), &request.CommonCountEventRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		KeyValues:             req.KeyValues,
		Version:               req.Version,
		Operation:             req.Operation,
		CreatorID:             req.CreatorID,
		StartCreatedTime:      req.StartCreatedTime,
		EndCreatedTime:        req.EndCreatedTime,
	})
	if err != nil {
		return "", 0, err
	}

	return reply.Statement, reply.Count, nil
}

func (c *Client) EventHistoryQueryByKeys(req *client.EventQueryByKeysRequest) (string, []client.EventInfo, int64, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.EventHistoryQueryByKeys(context.Background(), &request.EventQueryByKeysRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, 0, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), reply.TotalCount, nil
}

func (c *Client) CommonEventHistoryQuery(req *client.CommonEventQueryRequest) (string, []client.EventInfo, int64, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, 0, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.CommonEventHistoryQuery(context.Background(), &request.CommonEventQueryRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		Version:               req.Version,
		Operation:             req.Operation,
		CreatorID:             req.CreatorID,
		StartCreatedTime:      req.StartCreatedTime,
		EndCreatedTime:        req.EndCreatedTime,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, 0, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), reply.TotalCount, nil
}

func (c *Client) EventHistoryQueryOnlyByKeys(req *client.EventQueryByKeysRequest) (string, []client.EventInfo, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.EventHistoryQueryOnlyByKeys(context.Background(), &request.EventQueryByKeysRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), nil
}

func (c *Client) CommonEventHistoryQueryOnly(req *client.CommonEventQueryRequest) (string, []client.EventInfo, error) {
	var selectJsonBytes []byte
	if req.Select != nil {
		innerJsonBytes, err := req.Select.ToJson()
		if err != nil {
			return "", nil, err
		}

		selectJsonBytes = innerJsonBytes
	}

	reply, err := c.eventQueryServiceClient.CommonEventHistoryQueryOnly(context.Background(), &request.CommonEventQueryRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Select:                selectJsonBytes,
		KeyValues:             req.KeyValues,
		Version:               req.Version,
		Operation:             req.Operation,
		CreatorID:             req.CreatorID,
		StartCreatedTime:      req.StartCreatedTime,
		EndCreatedTime:        req.EndCreatedTime,
		PageNo:                int32(req.PageNo),
		PageSize:              int32(req.PageSize),
	})
	if err != nil {
		return "", nil, err
	}

	return reply.Statement, client.FormEventInfoBatch(reply.Infos), nil
}

func (c *Client) CountEventHistoryByKeys(req *client.CountEventByKeysRequest) (string, int64, error) {
	reply, err := c.eventQueryServiceClient.CountEventHistoryByKeys(context.Background(), &request.CountEventByKeysRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		KeyValues:             req.KeyValues,
	})
	if err != nil {
		return "", 0, err
	}

	return reply.Statement, reply.Count, nil
}

func (c *Client) CommonCountEventHistory(req *client.CommonCountEventRequest) (string, int64, error) {
	reply, err := c.eventQueryServiceClient.CommonCountEventHistory(context.Background(), &request.CommonCountEventRequest{
		DatabaseID:            c.databaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		KeyValues:             req.KeyValues,
		Version:               req.Version,
		Operation:             req.Operation,
		CreatorID:             req.CreatorID,
		StartCreatedTime:      req.StartCreatedTime,
		EndCreatedTime:        req.EndCreatedTime,
	})
	if err != nil {
		return "", 0, err
	}

	return reply.Statement, reply.Count, nil
}