shardservice

Documentation

Index

• Constants
• Variables
• type Config
  • func (c *Config) Validate()
• type Env
  • func NewEnv(service string, selectCNLabel string) Env
• type Option
  • func WithWaitCNReported() Option
• type ReadFunc
• type ReadOptions
  • func (opts ReadOptions) Adjust(fn func(*pb.TableShard)) ReadOptions
  • func (opts ReadOptions) PK(pk []byte) ReadOptions
  • func (opts ReadOptions) ReadAt(readAt timestamp.Timestamp) ReadOptions
  • func (opts ReadOptions) Shard(value uint64) ReadOptions
• type ReadRequest
• type ServerOption
• type ShardServer
  • func NewShardServer(cfg Config, logger *log.MOLogger, opts ...ServerOption) ShardServer
• type ShardService
  • func GetService(sid string) ShardService
  • func NewService(cfg Config, storage ShardStorage, opts ...Option) ShardService
• type ShardStorage
  • func NewShardStorage(sid string, clock clock.Clock, executor executor.SQLExecutor, ...) ShardStorage

Constants

const (
	ReadData                     = 0
	ReadRanges                   = 1
	ReadStats                    = 2
	ReadRows                     = 3
	ReadSize                     = 4
	ReadApproxObjectsNum         = 5
	ReadPrimaryKeysMayBeModified = 6
	ReadGetColumMetadataScanInfo = 7
	ReadBuildReader              = 8
	ReadMergeObjects             = 9
	ReadVisibleObjectStats       = 10
	ReadNext                     = 11
	ReadClose                    = 12
	ReadCollectTombstones        = 13
	ReadPrimaryKeysMayBeUpserted = 14
)

Variables

var (
	MetadataTableSQL = fmt.Sprintf(`create table %s.%s(
		table_id 		  bigint      unsigned primary key not null,
		account_id        bigint      unsigned not null,       
		policy            varchar(50)          not null,
		shard_count       int         unsigned not null,
		replica_count     int         unsigned not null,
		version           int         unsigned not null
	)`, catalog.MO_CATALOG, catalog.MOShardsMetadata)

	ShardsTableSQL = fmt.Sprintf(`create table %s.%s(
		table_id 		  bigint unsigned     not null,
		shard_id          bigint unsigned     not null,
		policy            varchar(50)         not null
	)`, catalog.MO_CATALOG, catalog.MOShards)

	InitSQLs = []string{
		MetadataTableSQL,
		ShardsTableSQL,
	}
)

var (
	DefaultOptions = ReadOptions{}
)

Types

type Config

type Config struct {
	// ServiceID service id
	ServiceID string `toml:"-"`
	// RPC rpc config
	RPC morpc.Config `toml:"-"`

	// Enable enable shard service
	Enable bool `toml:"enable"`
	// ListenAddress shard service listen address for receiving lock requests
	ListenAddress string `toml:"listen-address"`
	// ServiceAddress service address for communication, if this address is not set, use
	// ListenAddress as the communication address.
	ServiceAddress string `toml:"service-address"`
	// MaxScheduleTables how many tables's shards can be scheduled by balancer at once.
	MaxScheduleTables int `toml:"max-schedule-tables"`
	// FreezeCNTimeout max freeze time for CN can be scheduled by balancer again.
	FreezeCNTimeout toml.Duration `toml:"max-cn-freeze-time"`
	// ScheduleDuration how often to schedule shards on CNs.
	ScheduleDuration toml.Duration `toml:"schedule-duration"`
	// SelectCNLabel label name for tenant to select CNs.
	SelectCNLabel string `toml:"cn-select-label-name"`
	// HeartbeatDuration how often to send heartbeat to shard server.
	HeartbeatDuration toml.Duration `toml:"heartbeat-duration"`
	// CheckChangedDuration how often to check deleted tables or sharding changed in CN.
	CheckChangedDuration toml.Duration `toml:"check-changed-duration"`
}

Shard config

func (*Config) Validate

func (c *Config) Validate()

Validate validate

type Env

type Env interface {
	HasCN(serviceID string) bool
	Available(accountID uint64, cn string) bool
	Draining(cn string) bool

	UpdateState(cn string, state metadata.WorkState)
}

func NewEnv

func NewEnv(
	service string,
	selectCNLabel string,
) Env

type Option

type Option func(*service)

func WithWaitCNReported

func WithWaitCNReported() Option

type ReadFunc

type ReadFunc func(
	ctx context.Context,
	shard pb.TableShard,
	engine engine.Engine,
	param pb.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

type ReadOptions

type ReadOptions struct {
	// contains filtered or unexported fields
}

func (ReadOptions) Adjust

func (opts ReadOptions) Adjust(
	fn func(*pb.TableShard),
) ReadOptions

func (ReadOptions) PK

func (opts ReadOptions) PK(
	pk []byte,
) ReadOptions

func (ReadOptions) ReadAt

func (opts ReadOptions) ReadAt(
	readAt timestamp.Timestamp,
) ReadOptions

func (ReadOptions) Shard

func (opts ReadOptions) Shard(
	value uint64,
) ReadOptions

type ReadRequest

type ReadRequest struct {
	TableID uint64
	Method  int
	Param   pb.ReadParam
	Apply   func([]byte)
}

type ServerOption

type ServerOption func(*server)

type ShardServer

type ShardServer interface {
	// Close close the shard server
	Close() error
}

ShardServer used for balance and allocate shards on their cns. ShardServer adheres to the following principles:

1. As far as possible, it ensures that the table's shards are evenly distributed across the available CNs.
2. When an imbalance is found and the concurrency is readjusted, migrate the least number of shards to the new CN.
3. TableShard.BindVersion is incremented when the shard is bound to a new cn.

ShardServer periodically obtains information about the CNs in the cluster and performs a re-balance operation if it finds a change in the number of CNs.

The balancer generates a corresponding CMD when it creates a new bind, and 2 CMDs when it move a shard to another cn (one to delete the bind on the old cn, and one to add a new bind on the new cn). These CMDs are returned to the corresponding CMD for execution in the heartbeat request of each CN.

func NewShardServer

func NewShardServer(
	cfg Config,
	logger *log.MOLogger,
	opts ...ServerOption,
) ShardServer

type ShardService

type ShardService interface {
	// Config returns the configuration of the shard service.
	Config() Config
	// GetStorage returns the storage of the shard service.
	GetStorage() ShardStorage
	// Read read data from shards.
	Read(ctx context.Context, req ReadRequest, opts ReadOptions) error
	// HasLocalReplica returns whether the shard has a local replica.
	HasLocalReplica(tableID, shardID uint64) (bool, error)
	// HasAllLocalReplicas returns whether all shards of the table have local replicas.
	HasAllLocalReplicas(tableID uint64) (bool, error)
	// GetShardInfo returns the metadata of the shards corresponding to the table.
	GetShardInfo(table uint64) (shardTableID uint64, policy pb.Policy, isShardTable bool, err error)
	// GetShardAllocated returns the allocation of the shards corresponding to the table.
	GetTableShards(table uint64) (pb.ShardsMetadata, []pb.TableShard, error)
	// Create creates table shards metadata in current txn. And create shard
	// binds after txn committed asynchronously. Nothing happened if txn aborted.
	//
	// ShardBalancer will allocate CN after TableShardBind created.
	Create(ctx context.Context, table uint64, txnOp client.TxnOperator) error
	// Delete deletes table shards metadata in current txn. Table shards need
	// to be deleted if table deleted. Nothing happened if txn aborted.
	Delete(ctx context.Context, table uint64, txnOp client.TxnOperator) error
	// ReplicaCount returns the number of running replicas on current cn.
	ReplicaCount() int64
	// TableReplicaCount returns the number of running replicas of the special table on current cn.
	TableReplicaCount(tableID uint64) int64
	// Close close the service
	Close() error
}

ShardService is sharding service. Each CN node holds an instance of the ShardService.

func GetService

func GetService(
	sid string,
) ShardService

func NewService

func NewService(
	cfg Config,
	storage ShardStorage,
	opts ...Option,
) ShardService

type ShardStorage

type ShardStorage interface {
	// Get returns the latest metadata of the shards corresponding to the table.
	Get(table uint64) (uint64, pb.ShardsMetadata, error)
	// GetChanged returns the table ids of the shards that have been changed.
	GetChanged(tables map[uint64]uint32, applyDeleted func(uint64), applyChanged func(uint64)) error
	// Create creates the metadata for the sharding corresponding to the table with the given
	// transaction.
	Create(ctx context.Context, table uint64, txnOp client.TxnOperator) (bool, error)
	// Create delete the metadata for the sharding corresponding to the table with the given
	// transaction.
	Delete(ctx context.Context, table uint64, txnOp client.TxnOperator) (bool, error)
	// Unsubscribe unsubscribes the log tail of the tables.
	Unsubscribe(tables ...uint64) error
	// WaitLogAppliedAt wait until the log tail corresponding to ts has been fully consumed.
	// Ensure that subsequent reads have full log tail data.
	WaitLogAppliedAt(ctx context.Context, ts timestamp.Timestamp) error
	// Read read data with the given timestamp
	Read(ctx context.Context, shard pb.TableShard, method int, param pb.ReadParam, ts timestamp.Timestamp, buffer *morpc.Buffer) ([]byte, error)
}

ShardStorage is used to store metadata for Table Shards, handle read operations for shards, and Log tail subscriptions.

func NewShardStorage

func NewShardStorage(
	sid string,
	clock clock.Clock,
	executor executor.SQLExecutor,
	waiter client.TimestampWaiter,
	handles map[int]ReadFunc,
	engine engine.Engine,
) ShardStorage