merge

Documentation

Index

• type Merge
• type Merger
  • func NewMerger(db objects.Store, collector *RowCollector, buf *diff.BlockBuffer, ...) (m *Merger, err error)
  • func (m *Merger) Close() error
  • func (m *Merger) Columns(removedCols map[int]struct{}) []string
  • func (m *Merger) Error() error
  • func (m *Merger) PK() []string
  • func (m *Merger) SaveResolvedRow(pk []byte, row []string) error
  • func (m *Merger) SortedBlocks(removedCols map[int]struct{}) (<-chan *sorter.Block, uint32, error)
  • func (m *Merger) SortedRows(removedCols map[int]struct{}) (<-chan *sorter.Rows, uint32, error)
  • func (m *Merger) Start() (ch <-chan *Merge, err error)
• type RowCollector
  • func CreateRowCollector(db objects.Store, baseT *objects.Table) (collector *RowCollector, cleanup func(), err error)
  • func NewCollector(db objects.Store, baseT *objects.Table, discardedRow *index.HashSet) (*RowCollector, error)
  • func (c *RowCollector) Close() error
  • func (c *RowCollector) CollectResolvedRow(errChan chan<- error, origChan <-chan *Merge) <-chan *Merge
  • func (c *RowCollector) Columns(removedCols map[int]struct{}) []string
  • func (c *RowCollector) PK() []string
  • func (c *RowCollector) SaveResolvedRow(pk []byte, row []string) error
  • func (c *RowCollector) SortedBlocks(removedCols map[int]struct{}, errChan chan<- error) (<-chan *sorter.Block, uint32, error)
  • func (c *RowCollector) SortedRows(removedCols map[int]struct{}, errChan chan<- error) (<-chan *sorter.Rows, uint32, error)
• type RowResolver
  • func NewRowResolver(db objects.Store, cd *objects.ColDiff, buf *diff.BlockBuffer) *RowResolver
  • func (r *RowResolver) Resolve(m *Merge) (err error)
• type Rows
  • func NewRows(n int) *Rows
  • func (r *Rows) Append(layer int, row []string)
  • func (r *Rows) Len() int
  • func (r *Rows) Less(i, j int) bool
  • func (r *Rows) Reset()
  • func (r *Rows) Swap(i, j int)

Types

type Merge

type Merge struct {
	ColDiff        *objects.ColDiff
	PK             []byte
	Base           []byte
	BaseOffset     uint32
	Others         [][]byte
	OtherOffsets   []uint32
	ResolvedRow    []string
	Resolved       bool
	UnresolvedCols map[uint32]struct{}
}

type Merger

type Merger struct {
	Progress progress.Tracker
	// contains filtered or unexported fields
}

func NewMerger

func NewMerger(db objects.Store, collector *RowCollector, buf *diff.BlockBuffer, progressPeriod time.Duration, baseT *objects.Table, otherTs []*objects.Table, baseSum []byte, otherSums [][]byte) (m *Merger, err error)

func (*Merger) Close

func (m *Merger) Close() error

func (*Merger) Columns

func (m *Merger) Columns(removedCols map[int]struct{}) []string

func (*Merger) Error

func (m *Merger) Error() error

func (*Merger) PK

func (m *Merger) PK() []string

func (*Merger) SaveResolvedRow

func (m *Merger) SaveResolvedRow(pk []byte, row []string) error

func (*Merger) SortedBlocks

func (m *Merger) SortedBlocks(removedCols map[int]struct{}) (<-chan *sorter.Block, uint32, error)

func (*Merger) SortedRows

func (m *Merger) SortedRows(removedCols map[int]struct{}) (<-chan *sorter.Rows, uint32, error)

func (*Merger) Start

func (m *Merger) Start() (ch <-chan *Merge, err error)

type RowCollector

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

func CreateRowCollector

func CreateRowCollector(db objects.Store, baseT *objects.Table) (collector *RowCollector, cleanup func(), err error)

func NewCollector

func NewCollector(db objects.Store, baseT *objects.Table, discardedRow *index.HashSet) (*RowCollector, error)

func (*RowCollector) Close

func (c *RowCollector) Close() error

func (*RowCollector) CollectResolvedRow

func (c *RowCollector) CollectResolvedRow(errChan chan<- error, origChan <-chan *Merge) <-chan *Merge

func (*RowCollector) Columns

func (c *RowCollector) Columns(removedCols map[int]struct{}) []string

func (*RowCollector) PK

func (c *RowCollector) PK() []string

func (*RowCollector) SaveResolvedRow

func (c *RowCollector) SaveResolvedRow(pk []byte, row []string) error

func (*RowCollector) SortedBlocks

func (c *RowCollector) SortedBlocks(removedCols map[int]struct{}, errChan chan<- error) (<-chan *sorter.Block, uint32, error)

func (*RowCollector) SortedRows

func (c *RowCollector) SortedRows(removedCols map[int]struct{}, errChan chan<- error) (<-chan *sorter.Rows, uint32, error)

type RowResolver

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

func NewRowResolver

func NewRowResolver(db objects.Store, cd *objects.ColDiff, buf *diff.BlockBuffer) *RowResolver

func (*RowResolver) Resolve

func (r *RowResolver) Resolve(m *Merge) (err error)

type Rows

type Rows struct {
	Values [][]string
	Layers []int
}

func NewRows

func NewRows(n int) *Rows

func (*Rows) Append

func (r *Rows) Append(layer int, row []string)

func (*Rows) Len

func (r *Rows) Len() int

func (*Rows) Less

func (r *Rows) Less(i, j int) bool

func (*Rows) Reset

func (r *Rows) Reset()

func (*Rows) Swap

func (r *Rows) Swap(i, j int)