client

Documentation

Index

• type CachedSchemaHandler
  • func NewCachedSchemaHandler(logger *zap.SugaredLogger, scope tally.Scope, schamaFetcher SchemaFetcher) *CachedSchemaHandler
  • func (cf *CachedSchemaHandler) FetchAllSchema() error
  • func (cf *CachedSchemaHandler) FetchSchema(tableName string) (*TableSchema, error)
  • func (cf *CachedSchemaHandler) FetchSchemas()
  • func (cf *CachedSchemaHandler) PrepareEnumCases(tableName, columnName string, enumCases []string) error
  • func (cf *CachedSchemaHandler) Start(interval int)
  • func (cf *CachedSchemaHandler) TranslateEnum(tableName string, columnID int, value interface{}, caseInsensitive bool) (enumID int, err error)
• type Connector
• type ConnectorConfig
  • func (cfg ConnectorConfig) NewConnector(logger *zap.SugaredLogger, metricScope tally.Scope) Connector
• type Row
• type SchemaFetcher
  • func NewHttpSchemaFetcher(httpClient http.Client, address string, scope tally.Scope) SchemaFetcher
• type TableSchema
• type UpsertBatchBuilder
  • func NewUpsertBatchBuilderImpl(logger *zap.SugaredLogger, scope tally.Scope, ...) UpsertBatchBuilder
• type UpsertBatchBuilderImpl
  • func (u *UpsertBatchBuilderImpl) PrepareUpsertBatch(tableName string, columnNames []string, updateModes []memCom.ColumnUpdateMode, ...) ([]byte, int, error)

Types

type CachedSchemaHandler

type CachedSchemaHandler struct {
	*sync.RWMutex
	// contains filtered or unexported fields
}

CachedSchemaHandler handles schema and enum requests with cache

func NewCachedSchemaHandler

func NewCachedSchemaHandler(logger *zap.SugaredLogger, scope tally.Scope, schamaFetcher SchemaFetcher) *CachedSchemaHandler

NewCachedSchemaHandler creates a new cached schema handler

func (*CachedSchemaHandler) FetchAllSchema

func (cf *CachedSchemaHandler) FetchAllSchema() error

FetchAllSchema fetch all schemas

func (*CachedSchemaHandler) FetchSchema

func (cf *CachedSchemaHandler) FetchSchema(tableName string) (*TableSchema, error)

FetchSchema fetchs the schema of given table name

func (*CachedSchemaHandler) FetchSchemas

func (cf *CachedSchemaHandler) FetchSchemas()

FetchSchemas fetch schemas in schemas of CachedSchemaHandler

func (*CachedSchemaHandler) PrepareEnumCases

func (cf *CachedSchemaHandler) PrepareEnumCases(tableName, columnName string, enumCases []string) error

PrepareEnumCases prepares enum cases

func (*CachedSchemaHandler) Start

func (cf *CachedSchemaHandler) Start(interval int)

Start starts the CachedSchemaHandler, if interval > 0, will start periodical refresh

func (*CachedSchemaHandler) TranslateEnum

func (cf *CachedSchemaHandler) TranslateEnum(tableName string, columnID int, value interface{}, caseInsensitive bool) (enumID int, err error)

TranslateEnum translates given enum value to its enumID

type Connector

type Connector interface {
	// Insert inserts rows to ares
	// returns number of rows inserted and error.
	// updateModes are optional, if ignored for all columns, no need to set
	// if set, then all columns needs to be set
	Insert(tableName string, columnNames []string, rows []Row, updateModes ...memCom.ColumnUpdateMode) (int, error)
	// Close the connection
	Close()
}

Connector is the connector interface for ares.

type ConnectorConfig

type ConnectorConfig struct {
	// Address is in the format of host:port
	Address string `yaml:"address" json:"address"`
	// DeviceChoosingTimeout value is the request timeout in seconds for http calls
	// if <= 0, will use default
	Timeout int `yaml:"timeout" json:"timeout"`
	// SchemaRefreshInterval is the interval in seconds for the connector to
	// fetch and refresh schema from ares
	// if <= 0, will use default
	SchemaRefreshInterval int `yaml:"schemaRefreshInterval" json:"schemaRefreshInterval"`
}

ConnectorConfig holds the configurations for ares Connector.

func (ConnectorConfig) NewConnector

func (cfg ConnectorConfig) NewConnector(logger *zap.SugaredLogger, metricScope tally.Scope) Connector

NewConnector returns a new ares Connector

type Row

type Row []interface{}

Row represents a row of insert data.

type SchemaFetcher

type SchemaFetcher interface {
	// FetchAllSchemas fetches all schemas
	FetchAllSchemas() ([]*metaCom.Table, error)
	// FetchSchema fetch one schema for given table
	FetchSchema(table string) (*metaCom.Table, error)
	// FetchAllEnums fetches all enums for given table and column
	FetchAllEnums(tableName string, columnName string) ([]string, error)
	// ExtendEnumCases extends enum cases to given table column
	ExtendEnumCases(tableName, columnName string, enumCases []string) ([]int, error)
}

SchemaFetcher is the interface for fetch schema and enums

func NewHttpSchemaFetcher

func NewHttpSchemaFetcher(httpClient http.Client, address string, scope tally.Scope) SchemaFetcher

NewHttpSchemaFetcher creates a new http schema fetcher

type TableSchema

type TableSchema struct {
	Table *metaCom.Table
	// maps from column name to columnID for convenience
	ColumnDict map[string]int
}

type UpsertBatchBuilder

type UpsertBatchBuilder interface {
	PrepareUpsertBatch(tableName string, columnNames []string, updateModes []memCom.ColumnUpdateMode, rows []Row) ([]byte, int, error)
}

UpsertBatchBuilder is an interface of upsertBatch on client side

func NewUpsertBatchBuilderImpl

func NewUpsertBatchBuilderImpl(logger *zap.SugaredLogger, scope tally.Scope, schemaHandler *CachedSchemaHandler) UpsertBatchBuilder

type UpsertBatchBuilderImpl

type UpsertBatchBuilderImpl struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

UpsertBatchBuilderImpl implements interface UpsertBatchBuilder

func (*UpsertBatchBuilderImpl) PrepareUpsertBatch

func (u *UpsertBatchBuilderImpl) PrepareUpsertBatch(tableName string, columnNames []string,
	updateModes []memCom.ColumnUpdateMode, rows []Row) ([]byte, int, error)

PrepareUpsertBatch prepares the upsert batch for upsert, returns upsertBatch byte array, number of rows in upsert batch and error.