iterator

Documentation

Index

• Constants
• Variables
• type CDCIterator
  • func NewCDCIterator(snowflake Repository, table string, keys, columns []string, ...) *CDCIterator
  • func (c *CDCIterator) Ack(ctx context.Context, rp opencdc.Position) error
  • func (c *CDCIterator) HasNext(ctx context.Context) (bool, error)
  • func (c *CDCIterator) Next(_ context.Context) (opencdc.Record, error)
  • func (c *CDCIterator) Stop() error
• type Iterator
  • func New(ctx context.Context, conn, table, orderingColumn string, ...) (*Iterator, error)
  • func (i *Iterator) Ack(ctx context.Context, rp opencdc.Position) error
  • func (i *Iterator) HasNext(ctx context.Context) (bool, error)
  • func (i *Iterator) Next(ctx context.Context) (opencdc.Record, error)
  • func (i *Iterator) Stop() error
• type Repository

Constants

const (
	Conduit = "conduit"
)

Variables

var (
	ErrCantFindActionType       = errors.New("can't find action type")
	ErrInvalidSetup             = errors.New("invalid setup")
	ErrOrderingColumnIsNotExist = errors.New("ordering column is not exist")
)

Types

type CDCIterator

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

CDCIterator to iterate snowflake objects.

func NewCDCIterator

func NewCDCIterator(
	snowflake Repository,
	table string,
	keys, columns []string,
	index, offset, butchSize int,
	currentBatch []map[string]interface{},
) *CDCIterator

func (*CDCIterator) Ack

func (c *CDCIterator) Ack(ctx context.Context, rp opencdc.Position) error

Ack check if record with position was recorded.

func (*CDCIterator) HasNext

func (c *CDCIterator) HasNext(ctx context.Context) (bool, error)

HasNext check ability to get next record.

func (*CDCIterator) Next

func (c *CDCIterator) Next(_ context.Context) (opencdc.Record, error)

Next get new record.

func (*CDCIterator) Stop

func (c *CDCIterator) Stop() error

Stop shutdown iterator.

type Iterator

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

Iterator combined iterator.

func New

func New(
	ctx context.Context,
	conn, table,
	orderingColumn string,
	keys, columns []string,
	batchSize int,
	snapshot bool,
	pos opencdc.Position,
) (*Iterator, error)

New iterator.

func (*Iterator) Ack

func (i *Iterator) Ack(ctx context.Context, rp opencdc.Position) error

Ack check if record with position was recorded.

func (*Iterator) HasNext

func (i *Iterator) HasNext(ctx context.Context) (bool, error)

HasNext check ability to get next record.

func (*Iterator) Next

func (i *Iterator) Next(ctx context.Context) (opencdc.Record, error)

Next get new record.

func (*Iterator) Stop

func (i *Iterator) Stop() error

Stop iterator.

type Repository

type Repository interface {
	// GetRows - get rows from table.
	GetRows(ctx context.Context, table, orderingColumn string, fields []string, pos *position.Position, maxValue any,
		limit int) (*sqlx.Rows, error)
	// CreateStream - create stream.
	CreateStream(ctx context.Context, stream, table string) error
	// GetTrackingData - get rows from tracking table.
	GetTrackingData(ctx context.Context, stream, trackingTable string, fields []string,
		offset, limit int,
	) ([]map[string]interface{}, error)
	// CreateTrackingTable - create tracking table.
	CreateTrackingTable(ctx context.Context, trackingTable, table string) error
	// GetMaxValue get max value by ordering column.
	GetMaxValue(ctx context.Context, table, orderingColumn string) (any, error)
	// Close - shutdown repository.
	Close() error
	// GetPrimaryKeys returns all primary keys of the table.
	GetPrimaryKeys(ctx context.Context, table string) ([]string, error)
}

Repository interface.