stream

Documentation

Index

• Constants
• Variables
• type BaseTask
• type CacheRow
• type CacheRowPool
  • func NewCacheRowPool() *CacheRowPool
  • func (p *CacheRowPool) Get() *CacheRow
  • func (p *CacheRowPool) Len() int64
  • func (p *CacheRowPool) Put(r *CacheRow)
  • func (p *CacheRowPool) Size() int64
• type Engine
  • func NewStream(store Storage, Logger Logger, cli MetaClient, conf stream.Config) (Engine, error)
• type Logger
• type MetaClient
• type RowsPool
  • func NewRowsPool() *RowsPool
  • func (p *RowsPool) Get() *[]influx.Row
  • func (p *RowsPool) Put(r *[]influx.Row)
• type Storage
• type Stream
  • func (s *Stream) Close()
  • func (s *Stream) DeleteTask(id uint64)
  • func (s *Stream) Drain()
  • func (s *Stream) RegisterTask(info *meta2.StreamInfo, fieldCalls []*streamLib.FieldCall) error
  • func (s *Stream) Run()
  • func (s *Stream) WriteRows(db, rp string, ptId uint32, shardID uint64, ...)
• type TagTask
  • func (s *TagTask) Drain()
  • func (s *TagTask) Put(r *WindowCache)
  • func (s *TagTask) UnCompressDictKeyUint(key uint64) (string, error)
  • func (s *TagTask) WriteRowsToShard(start, end int) error
• type Task
• type TimeTask
  • func (s *TimeTask) Drain()
  • func (s *TimeTask) Put(r *WindowCache)
  • func (s *TimeTask) WriteRowsToShard() error
• type WindowCache
• type WindowCachePool
  • func NewWindowCachePool() *WindowCachePool
  • func (p *WindowCachePool) Count() int64
  • func (p *WindowCachePool) Get() *WindowCache
  • func (p *WindowCachePool) Put(r *WindowCache)
• type WindowDataPool
  • func NewWindowDataPool() *WindowDataPool
  • func (p *WindowDataPool) Get() *WindowCache
  • func (p *WindowDataPool) IncreaseChan()
  • func (p *WindowDataPool) Len() int64
  • func (p *WindowDataPool) Put(cache *WindowCache)
• type WritePointsWorkIF

Constants

const (
	EmptyGroupKey = ""
	EmptyTagValue = ""
)

Variables

var (
	FlushParallelMinRowNum = 10000
	ErrEmptyCache          = errors.New("empty window cache")
)

Types

type BaseTask

type BaseTask struct {
	Logger Logger
	// contains filtered or unexported fields
}

type CacheRow

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

type CacheRowPool

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

func NewCacheRowPool

func NewCacheRowPool() *CacheRowPool

func (*CacheRowPool) Get

func (p *CacheRowPool) Get() *CacheRow

func (*CacheRowPool) Len

func (p *CacheRowPool) Len() int64

func (*CacheRowPool) Put

func (p *CacheRowPool) Put(r *CacheRow)

func (*CacheRowPool) Size

func (p *CacheRowPool) Size() int64

type Engine

type Engine interface {
	WriteRows(db, rp string, ptId uint32, shardID uint64, streamIdDstShardIdMap map[uint64]uint64, ww WritePointsWorkIF)
	RegisterTask(info *meta2.StreamInfo, fieldCalls []*streamLib.FieldCall) error
	Drain()
	DeleteTask(id uint64)
	Run()
	Close()
}

func NewStream

func NewStream(store Storage, Logger Logger, cli MetaClient, conf stream.Config) (Engine, error)

type Logger

type Logger interface {
	Error(msg string, fields ...zap.Field)
	Info(msg string, fields ...zap.Field)
	Debug(msg string, fields ...zap.Field)
}

type MetaClient

type MetaClient interface {
	GetStreamInfosStore() map[string]*meta2.StreamInfo
	GetMeasurementInfoStore(dbName string, rpName string, mstName string) (*meta2.MeasurementInfo, error)
}

type RowsPool

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

func NewRowsPool

func NewRowsPool() *RowsPool

func (*RowsPool) Get

func (p *RowsPool) Get() *[]influx.Row

func (*RowsPool) Put

func (p *RowsPool) Put(r *[]influx.Row)

type Storage

type Storage interface {
	WriteRows(db, rp string, ptId uint32, shardID uint64, rows []influx.Row, binaryRows []byte) error
}

type Stream

type Stream struct {
	Logger Logger
	// contains filtered or unexported fields
}

func (*Stream) Close

func (s *Stream) Close()

Close shutdown

func (*Stream) DeleteTask

func (s *Stream) DeleteTask(id uint64)

func (*Stream) Drain

func (s *Stream) Drain()

Drain is for test case, to check whether there exist resource leakage

func (*Stream) RegisterTask

func (s *Stream) RegisterTask(info *meta2.StreamInfo, fieldCalls []*streamLib.FieldCall) error

func (*Stream) Run

func (s *Stream) Run()

func (*Stream) WriteRows

func (s *Stream) WriteRows(db, rp string, ptId uint32, shardID uint64, streamIdDstShardIdMap map[uint64]uint64,
	ww WritePointsWorkIF,
)

type TagTask

type TagTask struct {
	*WindowDataPool

	*BaseTask
	// contains filtered or unexported fields
}

func (*TagTask) Drain

func (s *TagTask) Drain()

func (*TagTask) Put

func (s *TagTask) Put(r *WindowCache)

func (*TagTask) UnCompressDictKeyUint

func (s *TagTask) UnCompressDictKeyUint(key uint64) (string, error)

UnCompressDictKeyUint corpusIndexes array not need lock

func (*TagTask) WriteRowsToShard

func (s *TagTask) WriteRowsToShard(start, end int) error

type Task

type Task interface {
	Drain()

	Put(r *WindowCache)
	// contains filtered or unexported methods
}

type TimeTask

type TimeTask struct {
	*WindowDataPool

	*BaseTask
	// contains filtered or unexported fields
}

func (*TimeTask) Drain

func (s *TimeTask) Drain()

func (*TimeTask) Put

func (s *TimeTask) Put(r *WindowCache)

func (*TimeTask) WriteRowsToShard

func (s *TimeTask) WriteRowsToShard() error

type WindowCache

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

type WindowCachePool

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

func NewWindowCachePool

func NewWindowCachePool() *WindowCachePool

func (*WindowCachePool) Count

func (p *WindowCachePool) Count() int64

func (*WindowCachePool) Get

func (p *WindowCachePool) Get() *WindowCache

func (*WindowCachePool) Put

func (p *WindowCachePool) Put(r *WindowCache)

type WindowDataPool

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

func NewWindowDataPool

func NewWindowDataPool() *WindowDataPool

func (*WindowDataPool) Get

func (p *WindowDataPool) Get() *WindowCache

func (*WindowDataPool) IncreaseChan

func (p *WindowDataPool) IncreaseChan()

func (*WindowDataPool) Len

func (p *WindowDataPool) Len() int64

func (*WindowDataPool) Put

func (p *WindowDataPool) Put(cache *WindowCache)

type WritePointsWorkIF

type WritePointsWorkIF interface {
	GetRows() []influx.Row
	PutWritePointsWork()
}