cteutil

Documentation

Index

• type Storage
• type StorageRC
  • func NewStorageRowContainer(tp []*types.FieldType, chkSize int) *StorageRC
  • func (s *StorageRC) ActionSpill() *chunk.SpillDiskAction
  • func (s *StorageRC) ActionSpillForTest() *chunk.SpillDiskAction
  • func (s *StorageRC) Add(chk *chunk.Chunk) (err error)
  • func (s *StorageRC) DerefAndClose() (err error)
  • func (s *StorageRC) Done() bool
  • func (s *StorageRC) Error() error
  • func (s *StorageRC) GetChunk(chkIdx int) (*chunk.Chunk, error)
  • func (s *StorageRC) GetDiskBytes() int64
  • func (s *StorageRC) GetDiskTracker() *memory.Tracker
  • func (s *StorageRC) GetIter() int
  • func (s *StorageRC) GetMemBytes() int64
  • func (s *StorageRC) GetMemTracker() *memory.Tracker
  • func (s *StorageRC) GetRow(ptr chunk.RowPtr) (chunk.Row, error)
  • func (s *StorageRC) Lock()
  • func (s *StorageRC) NumChunks() int
  • func (s *StorageRC) NumRows() int
  • func (s *StorageRC) OpenAndRef() (err error)
  • func (s *StorageRC) Reopen() (err error)
  • func (s *StorageRC) SetDone()
  • func (s *StorageRC) SetError(err error)
  • func (s *StorageRC) SetIter(iter int)
  • func (s *StorageRC) SwapData(other Storage) (err error)
  • func (s *StorageRC) Unlock()

Types

type Storage

type Storage interface {
	// If is first called, will open underlying storage. Otherwise will add ref count by one.
	OpenAndRef() error

	// Minus ref count by one, if ref count is zero, close underlying storage.
	DerefAndClose() (err error)

	// SwapData swaps data of two storage.
	// Other metainfo is not touched, such ref count/done flag etc.
	SwapData(other Storage) error

	// Reopen reset storage and related info.
	// So the status of Storage is like a new created one.
	Reopen() error

	// Add chunk into underlying storage.
	// Should return directly if chk is empty.
	Add(chk *chunk.Chunk) error

	// Get Chunk by index.
	GetChunk(chkIdx int) (*chunk.Chunk, error)

	// Get row by RowPtr.
	GetRow(ptr chunk.RowPtr) (chunk.Row, error)

	// NumChunks return chunk number of the underlying storage.
	NumChunks() int

	// NumRows return row number of the underlying storage.
	NumRows() int

	// Storage is not thread-safe.
	// By using Lock(), users can achieve the purpose of ensuring thread safety.
	Lock()
	Unlock()

	// Usually, Storage is filled first, then user can read it.
	// User can check whether Storage is filled first, if not, they can fill it.
	Done() bool
	SetDone()

	// Store error message, so we can return directly.
	Error() error
	SetError(err error)

	// Readers use iter information to determine
	// whether they need to read data from the beginning.
	SetIter(iter int)
	GetIter() int

	GetMemTracker() *memory.Tracker
	GetDiskTracker() *disk.Tracker
	ActionSpill() *chunk.SpillDiskAction

	GetMemBytes() int64
	GetDiskBytes() int64
}

Storage is a temporary storage to store the intermidate data of CTE.

Common usage as follows:

storage.Lock()
if !storage.Done() {
    fill all data into storage
}
storage.UnLock()
read data from storage

type StorageRC

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

StorageRC implements Storage interface using RowContainer.

func NewStorageRowContainer

func NewStorageRowContainer(tp []*types.FieldType, chkSize int) *StorageRC

NewStorageRowContainer create a new StorageRC.

func (*StorageRC) ActionSpill

func (s *StorageRC) ActionSpill() *chunk.SpillDiskAction

ActionSpill impls Storage ActionSpill interface.

func (*StorageRC) ActionSpillForTest

func (s *StorageRC) ActionSpillForTest() *chunk.SpillDiskAction

ActionSpillForTest is for test.

func (*StorageRC) Add

func (s *StorageRC) Add(chk *chunk.Chunk) (err error)

Add impls Storage Add interface.

func (*StorageRC) DerefAndClose

func (s *StorageRC) DerefAndClose() (err error)

DerefAndClose impls Storage DerefAndClose interface.

func (*StorageRC) Done

func (s *StorageRC) Done() bool

Done impls Storage Done interface.

func (*StorageRC) Error

func (s *StorageRC) Error() error

Error impls Storage Error interface.

func (*StorageRC) GetChunk

func (s *StorageRC) GetChunk(chkIdx int) (*chunk.Chunk, error)

GetChunk impls Storage GetChunk interface.

func (*StorageRC) GetDiskBytes

func (s *StorageRC) GetDiskBytes() int64

GetDiskBytes returns disk bytes used by row container.

func (*StorageRC) GetDiskTracker

func (s *StorageRC) GetDiskTracker() *memory.Tracker

GetDiskTracker impls Storage GetDiskTracker interface.

func (*StorageRC) GetIter

func (s *StorageRC) GetIter() int

GetIter impls Storage GetIter interface.

func (*StorageRC) GetMemBytes

func (s *StorageRC) GetMemBytes() int64

GetMemBytes returns memory bytes used by row container.

func (*StorageRC) GetMemTracker

func (s *StorageRC) GetMemTracker() *memory.Tracker

GetMemTracker impls Storage GetMemTracker interface.

func (*StorageRC) GetRow

func (s *StorageRC) GetRow(ptr chunk.RowPtr) (chunk.Row, error)

GetRow impls Storage GetRow interface.

func (*StorageRC) Lock

func (s *StorageRC) Lock()

Lock impls Storage Lock interface.

func (*StorageRC) NumChunks

func (s *StorageRC) NumChunks() int

NumChunks impls Storage NumChunks interface.

func (*StorageRC) NumRows

func (s *StorageRC) NumRows() int

NumRows impls Storage NumRows interface.

func (*StorageRC) OpenAndRef

func (s *StorageRC) OpenAndRef() (err error)

OpenAndRef impls Storage OpenAndRef interface.

func (*StorageRC) Reopen

func (s *StorageRC) Reopen() (err error)

Reopen impls Storage Reopen interface.

func (*StorageRC) SetDone

func (s *StorageRC) SetDone()

SetDone impls Storage SetDone interface.

func (*StorageRC) SetError

func (s *StorageRC) SetError(err error)

SetError impls Storage SetError interface.

func (*StorageRC) SetIter

func (s *StorageRC) SetIter(iter int)

SetIter impls Storage SetIter interface.

func (*StorageRC) SwapData

func (s *StorageRC) SwapData(other Storage) (err error)

SwapData impls Storage Swap interface.

func (*StorageRC) Unlock

func (s *StorageRC) Unlock()

Unlock impls Storage Unlock interface.