posting

Documentation

Overview

Package posting takes care of posting lists. It contains logic for mutation layers, merging them with RocksDB, etc.

Index

• Constants
• Variables
• func CommitLists(numRoutines int)
• func IndexTokens(attr string, src types.Val) ([]string, error)
• func Init(ps *store.Store)
• func KeysForTest(attr string) []string
• func StartCommit()
• func WaterMarkFor(group uint32) *x.WaterMark
• type ByUid
  • func (pa ByUid) Len() int
  • func (pa ByUid) Less(i, j int) bool
  • func (pa ByUid) Swap(i, j int)
• type List
  • func GetOrCreate(key []byte, group uint32) (rlist *List, decr func())
  • func (l *List) AddMutation(ctx context.Context, t *task.DirectedEdge) (bool, error)
  • func (l *List) AddMutationWithIndex(ctx context.Context, t *task.DirectedEdge) error
  • func (l *List) CommitIfDirty(ctx context.Context) (committed bool, err error)
  • func (l *List) Iterate(afterUid uint64, f func(obj *types.Posting) bool)
  • func (l *List) LastCompactionTs() time.Time
  • func (l *List) Length(afterUid uint64) int
  • func (l *List) PostingList() *types.PostingList
  • func (l *List) SetForDeletion()
  • func (l *List) Uids(opt ListOptions) *task.List
  • func (l *List) Value() (rval types.Val, rerr error)
  • func (l *List) WaitForCommit()
• type ListOptions
• type TokensTable
  • func GetTokensTable(attr string) *TokensTable
  • func NewTokensTable() *TokensTable
  • func (t *TokensTable) Add(s string)
  • func (t *TokensTable) Get(s string) int
  • func (t *TokensTable) GetFirst() string
  • func (t *TokensTable) GetLast() string
  • func (t *TokensTable) GetNext(key string) string
  • func (t *TokensTable) GetNextOrEqual(s string) string
  • func (t *TokensTable) GetPrev(key string) string
  • func (t *TokensTable) GetPrevOrEqual(s string) string
  • func (t *TokensTable) Size() int

Constants

const (
	// Set means overwrite in mutation layer. It contributes 0 in Length.
	Set uint32 = 0x01
	// Del means delete in mutation layer. It contributes -1 in Length.
	Del uint32 = 0x02
	// Add means add new element in mutation layer. It contributes 1 in Length.
	Add uint32 = 0x03
)

Variables

var (
	// ErrRetry can be triggered if the posting list got deleted from memory due to a hard commit.
	// In such a case, retry.
	ErrRetry = fmt.Errorf("Temporary Error. Please retry.")
	// ErrNoValue would be returned if no value was found in the posting list.
	ErrNoValue = fmt.Errorf("No value found")
)

Functions

func CommitLists

func CommitLists(numRoutines int)

func IndexTokens

func IndexTokens(attr string, src types.Val) ([]string, error)

IndexTokens return tokens, without the predicate prefix and index rune.

func Init

func Init(ps *store.Store)

Init initializes the posting lists package, the in memory and dirty list hash.

func KeysForTest

func KeysForTest(attr string) []string

KeysForTest returns keys for a table. This is just for testing / debugging.

func StartCommit

func StartCommit()

func WaterMarkFor

func WaterMarkFor(group uint32) *x.WaterMark

WaterMarkFor returns the synced watermark for the given RAFT group. We use this to determine the index to use when creating a new snapshot.

Types

type ByUid

type ByUid []*types.Posting

func (ByUid) Len

func (pa ByUid) Len() int

func (ByUid) Less

func (pa ByUid) Less(i, j int) bool

func (ByUid) Swap

func (pa ByUid) Swap(i, j int)

type List

type List struct {
	x.SafeMutex
	// contains filtered or unexported fields
}

func GetOrCreate

func GetOrCreate(key []byte, group uint32) (rlist *List, decr func())

GetOrCreate stores the List corresponding to key, if it's not there already. to lhmap and returns it. It also returns a reference decrement function to be called by caller.

plist, decr := GetOrCreate(key, store) defer decr() ... // Use plist TODO: This should take a node id and index. And just append all indices to a list. When doing a commit, it should update all the sync index watermarks. worker pkg would push the indices to the watermarks held by lists. And watermark stuff would have to be located outside worker pkg, maybe in x. That way, we don't have a dependency conflict.

func (*List) AddMutation

func (l *List) AddMutation(ctx context.Context, t *task.DirectedEdge) (bool, error)

AddMutation adds mutation to mutation layers. Note that it does not write anything to disk. Some other background routine will be responsible for merging changes in mutation layers to RocksDB. Returns whether any mutation happens.

func (*List) AddMutationWithIndex

func (l *List) AddMutationWithIndex(ctx context.Context, t *task.DirectedEdge) error

AddMutationWithIndex is AddMutation with support for indexing. It also supports reverse edges.

func (*List) CommitIfDirty

func (l *List) CommitIfDirty(ctx context.Context) (committed bool, err error)

func (*List) Iterate

func (l *List) Iterate(afterUid uint64, f func(obj *types.Posting) bool)

Iterate will allow you to iterate over this Posting List, while having acquired a read lock. So, please keep this iteration cheap, otherwise mutations would get stuck. The iteration will start after the provided UID. The results would not include this UID. The function will loop until either the Posting List is fully iterated, or you return a false in the provided function, which will indicate to the function to break out of the iteration.

	pl.Iterate(func(p *types.Posting) bool {
   // Use posting p
   return true  // to continue iteration.
   return false // to break iteration.
 })

func (*List) LastCompactionTs

func (l *List) LastCompactionTs() time.Time

func (*List) Length

func (l *List) Length(afterUid uint64) int

Length iterates over the mutation layer and counts number of elements.

func (*List) PostingList

func (l *List) PostingList() *types.PostingList

func (*List) SetForDeletion

func (l *List) SetForDeletion()

SetForDeletion will mark this List to be deleted, so no more mutations can be applied to this.

func (*List) Uids

func (l *List) Uids(opt ListOptions) *task.List

Uids returns the UIDs given some query params. We have to apply the filtering before applying (offset, count).

func (*List) Value

func (l *List) Value() (rval types.Val, rerr error)

func (*List) WaitForCommit

func (l *List) WaitForCommit()

type ListOptions

type ListOptions struct {
	AfterUID  uint64     // Any UID returned must be after this value.
	Intersect *task.List // Intersect results with this list of UIDs.
}

ListOptions is used in List.Uids (in posting) to customize our output list of UIDs, for each posting list. It should be internal to this package.

type TokensTable

type TokensTable struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

TokensTable tracks the keys / tokens / buckets for an indexed attribute.

func GetTokensTable

func GetTokensTable(attr string) *TokensTable

GetTokensTable returns TokensTable for an indexed attribute.

func NewTokensTable

func NewTokensTable() *TokensTable

NewTokensTable returns a new TokensTable.

func (*TokensTable) Add

func (t *TokensTable) Add(s string)

Add increments counter for a given key. If it doesn't exist, we create a new entry in TokensTable. We don't support delete yet. We are using a very simple implementation. In the future, as balanced trees / skip lists implementations become standardized for Go, we may consider using them. We also didn't support Delete operations yet. For that, we need to store the counts for each key.

func (*TokensTable) Get

func (t *TokensTable) Get(s string) int

Get returns position of element. If not found, it returns -1.

func (*TokensTable) GetFirst

func (t *TokensTable) GetFirst() string

GetFirst returns the first key in our list of keys. You could also call GetNext("") but that is less efficient.

func (*TokensTable) GetLast

func (t *TokensTable) GetLast() string

GetLast returns the first key in our list of keys. You could also call GetPrev("") but that is less efficient.

func (*TokensTable) GetNext

func (t *TokensTable) GetNext(key string) string

GetNext returns the next key after given key. If we reach the end, we return an empty string.

func (*TokensTable) GetNextOrEqual

func (t *TokensTable) GetNextOrEqual(s string) string

GetNextOrEqual returns position of leftmost element that is greater or equal to s.

func (*TokensTable) GetPrev

func (t *TokensTable) GetPrev(key string) string

GetPrev returns the key just before the given key. If we reach the start, we return an empty string.

func (*TokensTable) GetPrevOrEqual

func (t *TokensTable) GetPrevOrEqual(s string) string

GetPrevOrEqual returns position of rightmost element that is smaller or equal to s.

func (*TokensTable) Size

func (t *TokensTable) Size() int

Size returns size of TokensTable.