paging

Documentation

Overview

Package paging contains logic of splitting incoming data stream into the separate pages or blocks.

Index

• type Acceptor
• type ColumnarBuffer
• type ColumnarBufferFactory
  • func NewColumnarBufferFactory[T Acceptor](logger *zap.Logger, arrowAllocator memory.Allocator, ...) (ColumnarBufferFactory[T], error)
• type ColumnarBufferMock
  • func (m *ColumnarBufferMock) Release()
  • func (m *ColumnarBufferMock) ToResponse() (*api_service_protos.TReadSplitsResponse, error)
  • func (*ColumnarBufferMock) TotalRows() int
• type ReadLimiter
• type ReadLimiterFactory
  • func NewReadLimiterFactory(cfg *config.TServerReadLimit) *ReadLimiterFactory
  • func (rlf *ReadLimiterFactory) MakeReadLimiter(logger *zap.Logger) ReadLimiter
• type ReadResult
• type RowTransformer
  • func NewRowTransformer[T Acceptor](acceptors []T, appenders []func(acceptor T, builder array.Builder) error, ...) RowTransformer[T]
• type RowTransformerDefault
  • func (rt *RowTransformerDefault[T]) AppendToArrowBuilders(builders []array.Builder) error
  • func (rt *RowTransformerDefault[T]) GetAcceptors() []T
  • func (rt *RowTransformerDefault[T]) SetAcceptors(acceptors []T)
• type Sink
• type SinkFactory
  • func NewSinkFactory[T Acceptor](ctx context.Context, logger *zap.Logger, cfg *config.TPagingConfig, ...) SinkFactory[T]
• type SinkFactoryMock
  • func (m *SinkFactoryMock) FinalStats() *api_service_protos.TReadSplitsResponse_TStats
  • func (m *SinkFactoryMock) MakeSinks(totalSinks int) ([]Sink[any], error)
  • func (m *SinkFactoryMock) ResultQueue() <-chan *ReadResult[any]
• type SinkMock
  • func (m *SinkMock) AddError(err error)
  • func (m *SinkMock) AddRow(transformer RowTransformer[any]) error
  • func (m *SinkMock) Finish()
  • func (m *SinkMock) ResultQueue() <-chan *ReadResult[any]
• type TrafficTracker
  • func NewTrafficTracker[T Acceptor](pagination *config.TPagingConfig) *TrafficTracker[T]
  • func (tt *TrafficTracker[T]) DumpStats(total bool) *api_service_protos.TReadSplitsResponse_TStats

Types

type Acceptor

type Acceptor interface {
	any | string
}

Acceptor is a fundamental type class that is used during data extraction from the data source

type ColumnarBuffer

type ColumnarBuffer[T Acceptor] interface {

	// ToResponse returns all the accumulated data and clears buffer
	ToResponse() (*api_service_protos.TReadSplitsResponse, error)
	// Release frees resources if buffer is no longer used
	Release()
	// TotalRows return the number of rows accumulated
	TotalRows() int
	// contains filtered or unexported methods
}

type ColumnarBufferFactory

type ColumnarBufferFactory[T Acceptor] interface {
	MakeBuffer() (ColumnarBuffer[T], error)
}

func NewColumnarBufferFactory

func NewColumnarBufferFactory[T Acceptor](
	logger *zap.Logger,
	arrowAllocator memory.Allocator,
	format api_service_protos.TReadSplitsRequest_EFormat,
	selectWhat *api_service_protos.TSelect_TWhat,
) (ColumnarBufferFactory[T], error)

type ColumnarBufferMock

type ColumnarBufferMock struct {
	mock.Mock
}

func (*ColumnarBufferMock) Release

func (m *ColumnarBufferMock) Release()

func (*ColumnarBufferMock) ToResponse

func (m *ColumnarBufferMock) ToResponse() (*api_service_protos.TReadSplitsResponse, error)

func (*ColumnarBufferMock) TotalRows

func (*ColumnarBufferMock) TotalRows() int

type ReadLimiter

type ReadLimiter interface {
	// contains filtered or unexported methods
}

ReadLimiter helps to limitate amount of data returned by Connector server in every read request. This is generally should be avoided after https://st.yandex-team.ru/YQ-2057

type ReadLimiterFactory

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

func NewReadLimiterFactory

func NewReadLimiterFactory(cfg *config.TServerReadLimit) *ReadLimiterFactory

func (*ReadLimiterFactory) MakeReadLimiter

func (rlf *ReadLimiterFactory) MakeReadLimiter(logger *zap.Logger) ReadLimiter

type ReadResult

type ReadResult[T Acceptor] struct {
	ColumnarBuffer    ColumnarBuffer[T]
	Stats             *api_service_protos.TReadSplitsResponse_TStats
	Error             error
	IsTerminalMessage bool
}

ReadResult is an algebraic data type containing: 1. a buffer (e. g. page) packed with data 2. stats describing data that is kept in buffer 3. result of read operation (potentially with error) 4. flag marking this stream as completed

type RowTransformer

type RowTransformer[T Acceptor] interface {
	AppendToArrowBuilders(builders []array.Builder) error
	SetAcceptors(acceptors []T)
	GetAcceptors() []T
}

RowTransformer is a container for values taken extracted from a single table row. RowTransformer also knows how to convert them into columnar reprsentation with Arrow builders.

func NewRowTransformer

func NewRowTransformer[T Acceptor](
	acceptors []T,
	appenders []func(acceptor T, builder array.Builder) error,
	wantedColumnIDs []int,
) RowTransformer[T]

type RowTransformerDefault

type RowTransformerDefault[T Acceptor] struct {
	// contains filtered or unexported fields
}

func (*RowTransformerDefault[T]) AppendToArrowBuilders

func (rt *RowTransformerDefault[T]) AppendToArrowBuilders(builders []array.Builder) error

func (*RowTransformerDefault[T]) GetAcceptors

func (rt *RowTransformerDefault[T]) GetAcceptors() []T

func (*RowTransformerDefault[T]) SetAcceptors

func (rt *RowTransformerDefault[T]) SetAcceptors(acceptors []T)

type Sink

type Sink[T Acceptor] interface {
	// AddRow saves the row obtained from a stream incoming from an external data source.
	AddRow(rowTransformer RowTransformer[T]) error

	// Finish reports the successful completion of data stream reading.
	Finish()
}

Sink is a destination for a data stream that is read out of an external data source connection.

type SinkFactory

type SinkFactory[T Acceptor] interface {
	MakeSinks(totalSinks int) ([]Sink[T], error)
	// ResultQueue returns a channel with columnar buffers generated by all sinks
	ResultQueue() <-chan *ReadResult[T]
	// FinalStats returns the overall statistics collected during the request processing.
	FinalStats() *api_service_protos.TReadSplitsResponse_TStats
}

SinkFactory should be instantiated once for each ReadSplits request. It owns some structures that are shared across multiple Sink instances.

func NewSinkFactory

func NewSinkFactory[T Acceptor](
	ctx context.Context,
	logger *zap.Logger,
	cfg *config.TPagingConfig,
	columnarBufferFactory ColumnarBufferFactory[T],
	readLimiter ReadLimiter,
) SinkFactory[T]

type SinkFactoryMock

type SinkFactoryMock struct {
	mock.Mock
}

func (*SinkFactoryMock) FinalStats

func (m *SinkFactoryMock) FinalStats() *api_service_protos.TReadSplitsResponse_TStats

func (*SinkFactoryMock) MakeSinks

func (m *SinkFactoryMock) MakeSinks(totalSinks int) ([]Sink[any], error)

func (*SinkFactoryMock) ResultQueue

func (m *SinkFactoryMock) ResultQueue() <-chan *ReadResult[any]

type SinkMock

type SinkMock struct {
	mock.Mock
}

func (*SinkMock) AddError

func (m *SinkMock) AddError(err error)

func (*SinkMock) AddRow

func (m *SinkMock) AddRow(transformer RowTransformer[any]) error

func (*SinkMock) Finish

func (m *SinkMock) Finish()

func (*SinkMock) ResultQueue

func (m *SinkMock) ResultQueue() <-chan *ReadResult[any]

type TrafficTracker

type TrafficTracker[T Acceptor] struct {
	// contains filtered or unexported fields
}

func NewTrafficTracker

func NewTrafficTracker[T Acceptor](pagination *config.TPagingConfig) *TrafficTracker[T]

func (*TrafficTracker[T]) DumpStats

func (tt *TrafficTracker[T]) DumpStats(total bool) *api_service_protos.TReadSplitsResponse_TStats