craft

Documentation

Index

• Constants
• func DecodeTiDBType(ty byte, flag model.ColumnFlagType, bits []byte) (interface{}, error)
• func EncodeTiDBType(allocator *SliceAllocator, ty byte, flag model.ColumnFlagType, ...) []byte
• func NewBatchDecoderWithAllocator(allocator *SliceAllocator) codec.RowEventDecoder
• func NewBatchEncoder(config *common.Config) codec.RowEventEncoder
• func NewBatchEncoderBuilder(config *common.Config) codec.RowEventEncoderBuilder
• func NewBatchEncoderWithAllocator(allocator *SliceAllocator, config *common.Config) codec.RowEventEncoder
• type BatchEncoder
  • func (e *BatchEncoder) AppendRowChangedEvent(_ context.Context, _ string, ev *model.RowChangedEvent, callback func()) error
  • func (e *BatchEncoder) Build() []*common.Message
  • func (e *BatchEncoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error)
  • func (e *BatchEncoder) EncodeDDLEvent(ev *model.DDLEvent) (*common.Message, error)
• type Headers
  • func (h *Headers) Count() int
  • func (h *Headers) GetPartition(index int) int64
  • func (h *Headers) GetSchema(index int) string
  • func (h *Headers) GetTable(index int) string
  • func (h *Headers) GetTs(index int) uint64
  • func (h *Headers) GetType(index int) model.MessageType
• type MessageDecoder
  • func NewMessageDecoder(bits []byte, allocator *SliceAllocator) (*MessageDecoder, error)
  • func (d *MessageDecoder) DDLEvent(index int) (pmodel.ActionType, string, error)
  • func (d *MessageDecoder) Headers() (*Headers, error)
  • func (d *MessageDecoder) RowChangedEvent(index int) (preColumns, columns *columnGroup, err error)
• type MessageEncoder
  • func NewDDLEventEncoder(allocator *SliceAllocator, ev *model.DDLEvent) *MessageEncoder
  • func NewMessageEncoder(allocator *SliceAllocator) *MessageEncoder
  • func NewResolvedEventEncoder(allocator *SliceAllocator, ts uint64) *MessageEncoder
  • func (e *MessageEncoder) Encode() []byte
• type RowChangedEventBuffer
  • func NewRowChangedEventBuffer(allocator *SliceAllocator) *RowChangedEventBuffer
  • func (b *RowChangedEventBuffer) AppendRowChangedEvent(ev *model.RowChangedEvent, onlyHandleKeyColumns bool) (rows, size int)
  • func (b *RowChangedEventBuffer) Encode() []byte
  • func (b *RowChangedEventBuffer) GetHeaders() *Headers
  • func (b *RowChangedEventBuffer) Reset()
  • func (b *RowChangedEventBuffer) RowsCount() int
  • func (b *RowChangedEventBuffer) Size() int
• type SliceAllocator
  • func NewSliceAllocator(batchSize int) *SliceAllocator

Constants

const (
	// Version1 represents the version of craft format
	Version1 uint64 = 1

	// DefaultBufferCapacity is default buffer size
	DefaultBufferCapacity = 1024
)

Functions

func DecodeTiDBType

func DecodeTiDBType(ty byte, flag model.ColumnFlagType, bits []byte) (interface{}, error)

DecodeTiDBType decodes TiDB types.

func EncodeTiDBType

func EncodeTiDBType(allocator *SliceAllocator, ty byte, flag model.ColumnFlagType, value interface{}) []byte

EncodeTiDBType encodes TiDB types

func NewBatchDecoderWithAllocator

func NewBatchDecoderWithAllocator(
	allocator *SliceAllocator,
) codec.RowEventDecoder

NewBatchDecoderWithAllocator creates a new batchDecoder with given allocator.

func NewBatchEncoder

func NewBatchEncoder(config *common.Config) codec.RowEventEncoder

NewBatchEncoder creates a new BatchEncoder.

func NewBatchEncoderBuilder

func NewBatchEncoderBuilder(config *common.Config) codec.RowEventEncoderBuilder

NewBatchEncoderBuilder creates a craft batchEncoderBuilder.

func NewBatchEncoderWithAllocator

func NewBatchEncoderWithAllocator(allocator *SliceAllocator, config *common.Config) codec.RowEventEncoder

NewBatchEncoderWithAllocator creates a new BatchEncoder with given allocator.

Types

type BatchEncoder

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

BatchEncoder encodes the events into the byte of a batch into craft binary format.

func (*BatchEncoder) AppendRowChangedEvent

func (e *BatchEncoder) AppendRowChangedEvent(
	_ context.Context,
	_ string,
	ev *model.RowChangedEvent,
	callback func(),
) error

AppendRowChangedEvent implements the RowEventEncoder interface

func (*BatchEncoder) Build

func (e *BatchEncoder) Build() []*common.Message

Build implements the RowEventEncoder interface

func (*BatchEncoder) EncodeCheckpointEvent

func (e *BatchEncoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error)

EncodeCheckpointEvent implements the RowEventEncoder interface

func (*BatchEncoder) EncodeDDLEvent

func (e *BatchEncoder) EncodeDDLEvent(ev *model.DDLEvent) (*common.Message, error)

EncodeDDLEvent implements the RowEventEncoder interface

type Headers

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

Headers in columnar layout

func (*Headers) Count

func (h *Headers) Count() int

Count returns number of headers

func (*Headers) GetPartition

func (h *Headers) GetPartition(index int) int64

GetPartition returns partition of event at given index

func (*Headers) GetSchema

func (h *Headers) GetSchema(index int) string

GetSchema returns schema of event at given index

func (*Headers) GetTable

func (h *Headers) GetTable(index int) string

GetTable returns table of event at given index

func (*Headers) GetTs

func (h *Headers) GetTs(index int) uint64

GetTs returns timestamp of event at given index

func (*Headers) GetType

func (h *Headers) GetType(index int) model.MessageType

GetType returns type of event at given index

type MessageDecoder

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

MessageDecoder decoder

func NewMessageDecoder

func NewMessageDecoder(bits []byte, allocator *SliceAllocator) (*MessageDecoder, error)

NewMessageDecoder create a new message decode with bits and allocator

func (*MessageDecoder) DDLEvent

func (d *MessageDecoder) DDLEvent(index int) (pmodel.ActionType, string, error)

DDLEvent decode a DDL event

func (*MessageDecoder) Headers

func (d *MessageDecoder) Headers() (*Headers, error)

Headers decode headers of message

func (*MessageDecoder) RowChangedEvent

func (d *MessageDecoder) RowChangedEvent(index int) (preColumns, columns *columnGroup, err error)

RowChangedEvent decode a row changeded event

type MessageEncoder

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

MessageEncoder is encoder for message

func NewDDLEventEncoder

func NewDDLEventEncoder(allocator *SliceAllocator, ev *model.DDLEvent) *MessageEncoder

NewDDLEventEncoder creates a new encoder with given allocator and timestamp

func NewMessageEncoder

func NewMessageEncoder(allocator *SliceAllocator) *MessageEncoder

NewMessageEncoder creates a new encoder with given allocator

func NewResolvedEventEncoder

func NewResolvedEventEncoder(allocator *SliceAllocator, ts uint64) *MessageEncoder

NewResolvedEventEncoder creates a new encoder with given allocator and timestamp

func (*MessageEncoder) Encode

func (e *MessageEncoder) Encode() []byte

Encode message into bits

type RowChangedEventBuffer

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

RowChangedEventBuffer is a buffer to save row changed events in batch

func NewRowChangedEventBuffer

func NewRowChangedEventBuffer(allocator *SliceAllocator) *RowChangedEventBuffer

NewRowChangedEventBuffer creates new row changed event buffer with given allocator

func (*RowChangedEventBuffer) AppendRowChangedEvent

func (b *RowChangedEventBuffer) AppendRowChangedEvent(ev *model.RowChangedEvent, onlyHandleKeyColumns bool) (rows, size int)

AppendRowChangedEvent append a new event to buffer

func (*RowChangedEventBuffer) Encode

func (b *RowChangedEventBuffer) Encode() []byte

Encode row changed event buffer into bits

func (*RowChangedEventBuffer) GetHeaders

func (b *RowChangedEventBuffer) GetHeaders() *Headers

GetHeaders returns headers of buffer

func (*RowChangedEventBuffer) Reset

func (b *RowChangedEventBuffer) Reset()

Reset buffer

func (*RowChangedEventBuffer) RowsCount

func (b *RowChangedEventBuffer) RowsCount() int

RowsCount returns number of rows batched in this buffer.

func (*RowChangedEventBuffer) Size

func (b *RowChangedEventBuffer) Size() int

Size of buffer

type SliceAllocator

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

SliceAllocator for different slice types

func NewSliceAllocator

func NewSliceAllocator(batchSize int) *SliceAllocator

NewSliceAllocator creates a new slice allocator with given batch allocation size.