paging

Documentation

Overview

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

Index

• type ColumnarBuffer
• type ColumnarBufferFactory
  • func NewColumnarBufferFactory[T utils.Acceptor](logger log.Logger, arrowAllocator memory.Allocator, ...) (ColumnarBufferFactory[T], error)
• type ColumnarBufferMock
  • func (m *ColumnarBufferMock) Release()
  • func (m *ColumnarBufferMock) ToResponse() (*api_service_protos.TReadSplitsResponse, error)
  • func (m *ColumnarBufferMock) TotalRows() int
• type ReadLimiter
• type ReadLimiterFactory
  • func NewReadLimiterFactory(cfg *config.TServerReadLimit) *ReadLimiterFactory
  • func (rlf *ReadLimiterFactory) MakeReadLimiter(logger log.Logger) ReadLimiter
• type ReadResult
• type Sink
  • func NewSink[T utils.Acceptor](ctx context.Context, logger log.Logger, trafficTracker *TrafficTracker[T], ...) (Sink[T], error)
• type SinkMock
  • func (m *SinkMock) AddError(err error)
  • func (m *SinkMock) AddRow(transformer utils.RowTransformer[any]) error
  • func (m *SinkMock) Finish()
  • func (m *SinkMock) ResultQueue() <-chan *ReadResult[any]
• type TrafficTracker
  • func NewTrafficTracker[T utils.Acceptor](pagination *config.TPagingConfig) *TrafficTracker[T]
  • func (tt *TrafficTracker[T]) DumpStats(total bool) *api_service_protos.TReadSplitsResponse_TStats

Types

type ColumnarBuffer

type ColumnarBuffer[T utils.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 utils.Acceptor] interface {
	MakeBuffer() (ColumnarBuffer[T], error)
}

func NewColumnarBufferFactory

func NewColumnarBufferFactory[T utils.Acceptor](
	logger log.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 (m *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 log.Logger) ReadLimiter

type ReadResult

type ReadResult[T utils.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 Sink

type Sink[T utils.Acceptor] interface {
	// AddRow saves the row obtained from a stream incoming from an external data source.
	AddRow(rowTransformer utils.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 utils.Acceptor](
	ctx context.Context,
	logger log.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 utils.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 utils.Acceptor] struct {
	// contains filtered or unexported fields
}

func NewTrafficTracker

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

func (*TrafficTracker[T]) DumpStats

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