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
  • func NewSink[T Acceptor](ctx context.Context, logger *zap.Logger, trafficTracker *TrafficTracker[T], ...) (Sink[T], error)
• 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
}

ReadResult is an algebraic data type containing: 1. a buffer (e. g. page) packed with data 2. result of read operation (potentially with error)

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
	// AddError propagates an error occurred during the reading from the external data source.
	AddError(err error)
	// Finish reports the successful completion of reading the data stream.
	Finish()
	// ResultQueue returns a channel with results
	ResultQueue() <-chan *ReadResult[T]
}

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

func NewSink

func NewSink[T Acceptor](
	ctx context.Context,
	logger *zap.Logger,
	trafficTracker *TrafficTracker[T],
	columnarBufferFactory ColumnarBufferFactory[T],
	readLimiter ReadLimiter,
	resultQueueCapacity int,
) (Sink[T], error)

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