binlogreplication

README

The code in this directory was copied and modified from the Dolt project (as of 2024-08-28, https://github.com/dolthub/dolt/tree/main/go/libraries/doltcore/sqle/binlogreplication). The original code is licensed under the Apache License, Version 2.0. The modifications are also licensed under the Apache License, Version 2.0.

Documentation

Index

• Constants
• Variables
• type CommitKind
• type DeltaAppender
• type TableWriter
• type TableWriterProvider

Constants

const (
	ERNetReadError      = 1158
	ERFatalReplicaError = 13117
)

Variables

var ErrEmptyHostname = fmt.Errorf("fatal error: Invalid (empty) hostname when attempting to connect " +
	"to the source server. Connection attempt terminated")

ErrEmptyHostname is returned when replication is started without a hostname configured.

var ErrEmptyUsername = fmt.Errorf("fatal error: Invalid (empty) username when attempting to connect " +
	"to the source server. Connection attempt terminated")

ErrEmptyUsername is returned when replication is started without a username configured.

var ErrReplicationStopped = fmt.Errorf("replication stop requested")

ErrReplicationStopped is an internal error that is not returned to users, and signals that STOP REPLICA was called.

var ErrServerNotConfiguredAsReplica = fmt.Errorf(
	"server is not configured as a replica; fix with CHANGE REPLICATION SOURCE TO")

ErrServerNotConfiguredAsReplica is returned when replication is started without enough configuration provided.

var MyBinlogReplicaController = newMyBinlogReplicaController()

Types

type CommitKind

type CommitKind int

const (
	NormalCommit CommitKind = iota
	ImplicitCommitBeforeStmt
	ImplicitCommitAfterStmt
)

type DeltaAppender

type DeltaAppender interface {
	Field(i int) array.Builder
	Fields() []array.Builder
	Action() *array.Int8Builder
	TxnTag() *array.BinaryDictionaryBuilder
	TxnServer() *array.BinaryDictionaryBuilder
	TxnGroup() *array.BinaryDictionaryBuilder
	TxnSeqNumber() *array.Uint64Builder
	TxnStmtOrdinal() *array.Uint64Builder

	UpdateActionStats(action binlog.RowEventType, count int)
	ObserveEvents(event binlog.RowEventType, count int)
}

type TableWriter

type TableWriter interface {
	Insert(ctx *sql.Context, keyRows []sql.Row) error
	Delete(ctx *sql.Context, keyRows []sql.Row) error
	Update(ctx *sql.Context, keyRows []sql.Row, valueRows []sql.Row) error
}

type TableWriterProvider

type TableWriterProvider interface {
	// GetTableWriter returns a TableWriter for writing to the specified |table| in the specified |database|.
	GetTableWriter(
		ctx *sql.Context,
		txn *stdsql.Tx,
		databaseName, tableName string,
		schema sql.PrimaryKeySchema,
		columnCount, rowCount int,
		identifyColumns, dataColumns mysql.Bitmap,
		eventType binlog.RowEventType,
		foreignKeyChecksDisabled bool,
	) (TableWriter, error)

	// GetDeltaAppender returns a DeltaAppender for appending updates to the specified |table| in the specified |database|.
	GetDeltaAppender(
		ctx *sql.Context,
		databaseName, tableName string,
		schema sql.Schema,
	) (DeltaAppender, error)

	// FlushDelta writes the accumulated changes to the database.
	FlushDeltaBuffer(ctx *sql.Context, conn *stdsql.Conn, tx *stdsql.Tx, reason delta.FlushReason) error

	// DiscardDeltaBuffer discards the accumulated changes.
	DiscardDeltaBuffer(ctx *sql.Context)
}