vstreamer

Documentation

Index

• Constants
• Variables
• type ColExpr
• type CopyState
• type Engine
  • func NewEngine(env tabletenv.Env, ts srvtopo.Server, se *schema.Engine, ...) *Engine
  • func (vse *Engine) Close()
  • func (vse *Engine) InitDBConfig(keyspace string)
  • func (vse *Engine) IsOpen() bool
  • func (vse *Engine) Open()
  • func (vse *Engine) ServeHTTP(response http.ResponseWriter, request *http.Request)
  • func (vse *Engine) Stream(ctx context.Context, startPos string, tablePKs []*binlogdatapb.TableLastPK, ...) error
  • func (vse *Engine) StreamResults(ctx context.Context, query string, ...) error
  • func (vse *Engine) StreamRows(ctx context.Context, query string, lastpk []sqltypes.Value, ...) error
• type Filter
• type Opcode
• type Plan
• type RowStreamer
  • func NewRowStreamer(ctx context.Context, cp dbconfigs.Connector, se *schema.Engine, query string, ...) RowStreamer
• type Table

Constants

const (
	// Equal is used to filter an integer column on a specific value
	Equal = Opcode(iota)
	// VindexMatch is used for an in_keyrange() construct
	VindexMatch
)

Variables

var HeartbeatTime = 900 * time.Millisecond

HeartbeatTime is set to slightly below 1s, compared to idleTimeout set by VPlayer at slightly above 1s. This minimizes conflicts between the two timeouts.

var PacketSize = flag.Int("vstream_packet_size", 250000, "Suggested packet size for VReplication streamer. This is used only as a recommendation. The actual packet size may be more or less than this amount.")

PacketSize is the suggested packet size for VReplication streamer.

Types

type ColExpr

type ColExpr struct {
	// ColNum specifies the source column value.
	ColNum int

	// Vindex and VindexColumns, if set, will be used to generate
	// a keyspace_id. If so, ColNum is ignored.
	// VindexColumns contains the column numbers of the table,
	// and not the column numbers of the stream to be sent.
	Vindex        vindexes.Vindex
	VindexColumns []int

	Field *querypb.Field

	FixedValue sqltypes.Value
}

ColExpr represents a column expression.

type CopyState

type CopyState map[string][]*sqltypes.Result

CopyState contains the last PK for tables to be copied

type Engine

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

Engine is the engine for handling vreplication streaming requests.

func NewEngine

func NewEngine(env tabletenv.Env, ts srvtopo.Server, se *schema.Engine, lagThrottler *throttle.Throttler, cell string) *Engine

NewEngine creates a new Engine. Initialization sequence is: NewEngine->InitDBConfig->Open. Open and Close can be called multiple times and are idempotent.

func (*Engine) Close

func (vse *Engine) Close()

Close closes the Engine service.

func (*Engine) InitDBConfig

func (vse *Engine) InitDBConfig(keyspace string)

InitDBConfig initializes the target parameters for the Engine.

func (*Engine) IsOpen

func (vse *Engine) IsOpen() bool

IsOpen checks if the engine is opened

func (*Engine) Open

func (vse *Engine) Open()

Open starts the Engine service.

func (*Engine) ServeHTTP

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

ServeHTTP shows the current VSchema.

func (*Engine) Stream

func (vse *Engine) Stream(ctx context.Context, startPos string, tablePKs []*binlogdatapb.TableLastPK, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error

Stream starts a new stream. This streams events from the binary logs

func (*Engine) StreamResults

func (vse *Engine) StreamResults(ctx context.Context, query string, send func(*binlogdatapb.VStreamResultsResponse) error) error

StreamResults streams results of the query with the gtid.

func (*Engine) StreamRows

func (vse *Engine) StreamRows(ctx context.Context, query string, lastpk []sqltypes.Value, send func(*binlogdatapb.VStreamRowsResponse) error) error

StreamRows streams rows. This streams the table data rows (so we can copy the table data snapshot)

type Filter

type Filter struct {
	Opcode Opcode
	ColNum int
	Value  sqltypes.Value

	// Parameters for VindexMatch.
	// Vindex, VindexColumns and KeyRange, if set, will be used
	// to filter the row.
	// VindexColumns contains the column numbers of the table,
	// and not the column numbers of the stream to be sent.
	Vindex        vindexes.Vindex
	VindexColumns []int
	KeyRange      *topodatapb.KeyRange
}

Filter contains opcodes for filtering.

type Opcode

type Opcode int

Opcode enumerates the operators supported in a where clause

type Plan

type Plan struct {
	Table *Table

	// ColExprs is the list of column expressions to be sent
	// in the stream.
	ColExprs []ColExpr

	// Filters is the list of filters to be applied to the columns
	// of the table.
	Filters []Filter
}

Plan represents the plan for a table.

type RowStreamer

type RowStreamer interface {
	Stream() error
	Cancel()
}

RowStreamer exposes an externally usable interface to rowStreamer.

func NewRowStreamer

func NewRowStreamer(ctx context.Context, cp dbconfigs.Connector, se *schema.Engine, query string, lastpk []sqltypes.Value, send func(*binlogdatapb.VStreamRowsResponse) error, vse *Engine) RowStreamer

NewRowStreamer returns a RowStreamer

type Table

type Table struct {
	Name   string
	Fields []*querypb.Field
}

Table contains the metadata for a table.