utils

Documentation

Index

• Variables
• func RangeOverlapped(r1 *pb.Task_Range, r2 *pb.Task_Range) bool
• func ReverseBytesToTs(b []byte) uint64
• func SortDedupSlice(src [][]byte) [][]byte
• func TsRangeOverlapped(low1, high1, low2, high2 uint64) bool
• func TsToReverseBytes(ts uint64) []byte
• type ApiServerConfig
• type DBPool
  • func NewDBPool(addr string, port int, user, password, dbName string) (*DBPool, error)
• type DispatcherConfig
• type GlobalConfig
• type GrpcServer
  • func NewGrpcServer(addr string, port int) *GrpcServer
  • func (s *GrpcServer) Start(eg *errgroup.Group, ctx context.Context) error
  • func (s *GrpcServer) Stop() error
• type MetaServerConfig
• type SchemaRegistryConfig
• type SorterConfig
• type StorageConfig

Variables

var (
	ErrContentTooShort                       = fmt.Errorf("content too short")
	ErrChecksumNotMatch                      = fmt.Errorf("checksum not match")
	ErrInvalidSchemaSnapFile                 = fmt.Errorf("invalid schema snap file")
	ErrInvalidDataChangeFile                 = fmt.Errorf("invalid data change file")
	ErrInvalidDataChangeFileChecksum         = fmt.Errorf("invalid data change file checksum")
	ErrInvalidSchemaSnapFileChecksum         = fmt.Errorf("invalid schema snap file checksum")
	ErrTenantAlreadyAttched                  = fmt.Errorf("tenant already attached")
	ErrInvalidSorterBufferSnapFile           = fmt.Errorf("invalid sorter buffer snap file")
	ErrInvalidSorterBufferSnapFileChecksum   = fmt.Errorf("invalid sorter buffer snap file checksum")
	ErrUnmatchedTenantID                     = fmt.Errorf("unmatched tenant id")
	ErrUnmatchedRange                        = fmt.Errorf("unmatched range")
	ErrInvalidPrewriteBufferSnapFile         = fmt.Errorf("invalid prewrite buffer snap file")
	ErrInvalidPrewriteBufferSnapFileChecksum = fmt.Errorf("invalid prewrite buffer snap file checksum")
	ErrInvalidSorterBufferSnapFileVersion    = fmt.Errorf("invalid sorter buffer snap file version")
	ErrInvalidPrewriteBufferSnapFileVersion  = fmt.Errorf("invalid prewrite buffer snap file version")
	ErrInvalidRange                          = fmt.Errorf("invalid range")
	ErrInvalidRangeWatermarksSnapFile        = fmt.Errorf("invalid range watermarks snap file")
	ErrInvalidRangeWatermarksSnapFileVersion = fmt.Errorf("invalid range watermarks snap file version")
	ErrInvalidRangeWatermarksSnapTenantID    = fmt.Errorf("invalid range watermarks snap tenant id")
	ErrInvalidRangeWatermarksSnapRange       = fmt.Errorf("invalid range watermarks snap range")
	ErrInvalidCommittedDataFile              = fmt.Errorf("invalid committed data file")
	ErrInvalidCommittedDataFileVersion       = fmt.Errorf("invalid committed data file version")
	ErrInvalidCommittedManifestFile          = fmt.Errorf("invalid committed manifest file")
	ErrInvalidCommittedRangeIndexFile        = fmt.Errorf("invalid committed range index file")
)

Functions

func RangeOverlapped

func RangeOverlapped(r1 *pb.Task_Range, r2 *pb.Task_Range) bool

func ReverseBytesToTs

func ReverseBytesToTs(b []byte) uint64

func SortDedupSlice

func SortDedupSlice(src [][]byte) [][]byte

func TsRangeOverlapped

func TsRangeOverlapped(low1, high1, low2, high2 uint64) bool

func TsToReverseBytes

func TsToReverseBytes(ts uint64) []byte

Types

type ApiServerConfig

type ApiServerConfig struct {
	Addr    string
	Port    int
	Storage StorageConfig
}

type DBPool

type DBPool struct {
	*sql.DB
}

DBPool is database connection pool

func NewDBPool

func NewDBPool(addr string, port int, user, password, dbName string) (*DBPool, error)

NewDBPool creates a new database connection pool.

type DispatcherConfig

type DispatcherConfig struct {
	Addr string
	Port int

	MetaServerEndpoints []string

	// If the dispatcher is isolated from the meta server for a long time,
	// it will be paused.
	SuicideDur int

	// storage configuration
	Storage StorageConfig

	// How often to check the store changes, in seconds.
	// Store changes caused by upstream scale out or scale in.
	// Default to 10 seconds.
	CheckStoreInterval int
	// How often to check the new file changes, in seconds.
	// Default to 5 seconds.
	CheckFileInterval int
}

type GlobalConfig

type GlobalConfig struct {
	LogLevel   string
	Dispatcher DispatcherConfig
	Sorter     SorterConfig
	MetaServer MetaServerConfig
	ApiServer  ApiServerConfig
}

GlobalConfig is the global configuration of the system. Please refer to etc/config.toml for the details of each field.

type GrpcServer

type GrpcServer struct {

	// The internal grpc server.
	InternalServer *grpc.Server
	// contains filtered or unexported fields
}

GrpcServer is the generic tcp server which provide protobuf rpc service.

func NewGrpcServer

func NewGrpcServer(addr string, port int) *GrpcServer

NewGrpcServer creates a new TcpServer.

func (*GrpcServer) Start

func (s *GrpcServer) Start(eg *errgroup.Group, ctx context.Context) error

Start starts the server.

func (*GrpcServer) Stop

func (s *GrpcServer) Stop() error

Stop stops the server.

type MetaServerConfig

type MetaServerConfig struct {
	Addr          string
	Port          int
	MysqlHost     string
	MysqlPort     int
	MysqlUser     string
	MysqlPassword string

	// The interval to update the lease of the master role, in seconds.
	// Default to 5 seconds.
	UpdateLeaseInterval int
	// Lease duration, in seconds, default to 10 seconds.
	LeaseDuration int
}

type SchemaRegistryConfig

type SchemaRegistryConfig struct {
	Addr string
	Port int
}

type SorterConfig

type SorterConfig struct {
	Addr    string
	Port    int
	Storage StorageConfig
}

type StorageConfig

type StorageConfig struct {
	Region      string
	Bucket      string
	Prefix      string
	SecurityKey string
	AccessKey   string
}