schema

Documentation

Index

• Constants
• Variables
• type Engine
  • func NewEngine(checker connpool.MySQLChecker, config tabletenv.TabletConfig) *Engine
  • func NewEngineForTests() *Engine
  • func (se *Engine) Close()
  • func (se *Engine) GetSchema() map[string]*Table
  • func (se *Engine) GetTable(tableName sqlparser.TableIdent) *Table
  • func (se *Engine) InitDBConfig(cp dbconfigs.Connector)
  • func (se *Engine) IsOpen() bool
  • func (se *Engine) MakeNonMaster()
  • func (se *Engine) Open() error
  • func (se *Engine) RegisterNotifier(name string, f notifier)
  • func (se *Engine) Reload(ctx context.Context) error
  • func (se *Engine) ServeHTTP(response http.ResponseWriter, request *http.Request)
  • func (se *Engine) SetTableForTests(table *Table)
  • func (se *Engine) UnregisterNotifier(name string)
• type MessageInfo
• type SequenceInfo
• type Table
  • func LoadTable(conn *connpool.DBConn, tableName string, tableType string, comment string) (*Table, error)
  • func NewTable(name string) *Table
  • func (ta *Table) FindColumn(name sqlparser.ColIdent) int
  • func (ta *Table) GetPKColumn(index int) *querypb.Field
  • func (ta *Table) HasPrimary() bool

Constants

const (
	NoType = iota
	Sequence
	Message
)

Table types

Variables

var TypeNames = []string{
	"none",
	"sequence",
	"message",
}

TypeNames allows to fetch a the type name for a table. Count must match the number of table types.

Types

type Engine

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

Engine stores the schema info and performs operations that keep itself up-to-date.

func NewEngine

func NewEngine(checker connpool.MySQLChecker, config tabletenv.TabletConfig) *Engine

NewEngine creates a new Engine.

func NewEngineForTests

func NewEngineForTests() *Engine

NewEngineForTests creates a new engine, that can't query the database, and will not send notifications. It starts opened, and doesn't reload. Use SetTableForTests to set table schema.

func (*Engine) Close

func (se *Engine) Close()

Close shuts down Engine and is idempotent. It can be re-opened after Close.

func (*Engine) GetSchema

func (se *Engine) GetSchema() map[string]*Table

GetSchema returns the current The Tables are a shared data structure and must be treated as read-only.

func (*Engine) GetTable

func (se *Engine) GetTable(tableName sqlparser.TableIdent) *Table

GetTable returns the info for a table.

func (*Engine) InitDBConfig

func (se *Engine) InitDBConfig(cp dbconfigs.Connector)

InitDBConfig must be called before Open.

func (*Engine) IsOpen

func (se *Engine) IsOpen() bool

IsOpen checks if engine is open

func (*Engine) MakeNonMaster

func (se *Engine) MakeNonMaster()

MakeNonMaster clears the sequence caches to make sure that they don't get accidentally reused after losing mastership.

func (*Engine) Open

func (se *Engine) Open() error

Open initializes the Engine. Calling Open on an already open engine is a no-op.

func (*Engine) RegisterNotifier

func (se *Engine) RegisterNotifier(name string, f notifier)

RegisterNotifier registers the function for schema change notification. It also causes an immediate notification to the caller. The notified function must not change the map or its contents. The only exception is the sequence table where the values can be changed using the lock.

func (*Engine) Reload

func (se *Engine) Reload(ctx context.Context) error

Reload reloads the schema info from the db. Any tables that have changed since the last load are updated.

func (*Engine) ServeHTTP

func (se *Engine) ServeHTTP(response http.ResponseWriter, request *http.Request)

func (*Engine) SetTableForTests

func (se *Engine) SetTableForTests(table *Table)

SetTableForTests puts a Table in the map directly.

func (*Engine) UnregisterNotifier

func (se *Engine) UnregisterNotifier(name string)

UnregisterNotifier unregisters the notifier function.

type MessageInfo

type MessageInfo struct {
	// Fields stores the field info to be
	// returned for subscribers.
	Fields []*querypb.Field

	// AckWaitDuration specifies how long to wait after
	// the message was first sent. The back-off doubles
	// every attempt.
	AckWaitDuration time.Duration

	// PurgeAfterDuration specifies the time after which
	// a successfully acked message can be deleted.
	PurgeAfterDuration time.Duration

	// BatchSize specifies the max number of events to
	// send per response.
	BatchSize int

	// CacheSize specifies the number of messages to keep
	// in cache. Anything that cannot fit in the cache
	// is sent as best effort.
	CacheSize int

	// PollInterval specifies the polling frequency to
	// look for messages to be sent.
	PollInterval time.Duration
}

MessageInfo contains info specific to message tables.

type SequenceInfo

type SequenceInfo struct {
	sync.Mutex
	NextVal int64
	LastVal int64
}

SequenceInfo contains info specific to sequence tabels. It must be locked before accessing the values inside. If CurVal==LastVal, we have to cache new values. When the schema is first loaded, the values are all 0, which will trigger caching on first use.

type Table

type Table struct {
	Name      sqlparser.TableIdent
	Fields    []*querypb.Field
	PKColumns []int
	Type      int

	// SequenceInfo contains info for sequence tables.
	SequenceInfo *SequenceInfo

	// MessageInfo contains info for message tables.
	MessageInfo *MessageInfo
}

Table contains info about a table.

func LoadTable

func LoadTable(conn *connpool.DBConn, tableName string, tableType string, comment string) (*Table, error)

LoadTable creates a Table from the schema info in the database.

func NewTable

func NewTable(name string) *Table

NewTable creates a new Table.

func (*Table) FindColumn

func (ta *Table) FindColumn(name sqlparser.ColIdent) int

FindColumn finds a column in the table. It returns the index if found. Otherwise, it returns -1.

func (*Table) GetPKColumn

func (ta *Table) GetPKColumn(index int) *querypb.Field

GetPKColumn returns the pk column specified by the index.

func (*Table) HasPrimary

func (ta *Table) HasPrimary() bool

HasPrimary returns true if the table has a primary key.