jldb/mdb/pfile/index.go

package pfile

import (
	"git.crumpington.com/public/jldb/lib/errs"
	"git.crumpington.com/public/jldb/mdb/change"
)

type Index struct {
	fList *freeList
	aList allocList
	seen  map[[2]uint64]struct{}
	mask  []bool
}

func NewIndex(f *File) (*Index, error) {
	firstPage, err := f.pageCount()
	if err != nil {
		return nil, err
	}

	idx := &Index{
		fList: newFreeList(firstPage),
		aList: *newAllocList(),
		seen:  map[[2]uint64]struct{}{},
		mask:  []bool{},
	}

	err = f.iterate(func(pageID uint64, page dataPage) error {
		header := page.Header()
		switch header.PageType {
		case pageTypeHead:
			idx.aList.Create(header.CollectionID, header.ItemID, pageID)
		case pageTypeData:
			if !idx.aList.Push(header.CollectionID, header.ItemID, pageID) {
				return errs.Corrupt.WithMsg("encountered data page with no corresponding head page")
			}
		case pageTypeFree:
			idx.fList.Push(pageID)
		}
		return nil
	})

	return idx, err
}

func (idx *Index) StageChanges(changes []change.Change) {
	clear(idx.seen)
	if cap(idx.mask) < len(changes) {
		idx.mask = make([]bool, len(changes))
	}
	idx.mask = idx.mask[:len(changes)]

	for i := len(changes) - 1; i >= 0; i-- {
		key := [2]uint64{changes[i].CollectionID, changes[i].ItemID}
		if _, ok := idx.seen[key]; ok {
			idx.mask[i] = false
			continue
		}

		idx.seen[key] = struct{}{}
		idx.mask[i] = true
	}

	for i, active := range idx.mask {
		if !active {
			continue
		}

		if changes[i].Store {
			count := idx.getPageCountForData(len(changes[i].Data))
			changes[i].WritePageIDs = idx.fList.Pop(count, changes[i].WritePageIDs)
		}

		if pages := idx.aList.Remove(changes[i].CollectionID, changes[i].ItemID); pages != nil {
			changes[i].ClearPageIDs = pages
		}
	}
}

func (idx *Index) UnstageChanges(changes []change.Change) {
	for i := range changes {
		if len(changes[i].WritePageIDs) > 0 {
			idx.fList.Push(changes[i].WritePageIDs...)
			changes[i].WritePageIDs = changes[i].WritePageIDs[:0]
		}
		if len(changes[i].ClearPageIDs) > 0 {
			idx.aList.Store(changes[i].CollectionID, changes[i].ItemID, changes[i].ClearPageIDs)
			changes[i].ClearPageIDs = changes[i].ClearPageIDs[:0]
		}
	}
}

func (idx *Index) ApplyChanges(changes []change.Change) {
	for i := range changes {
		if len(changes[i].WritePageIDs) > 0 {
			idx.aList.Store(changes[i].CollectionID, changes[i].ItemID, changes[i].WritePageIDs)
		}
		if len(changes[i].ClearPageIDs) > 0 {
			idx.fList.Push(changes[i].ClearPageIDs...)
		}
	}
}

func (idx *Index) getPageCountForData(dataSize int) int {
	count := dataSize / pageDataSize
	if dataSize%pageDataSize != 0 {
		count++
	}
	return count
}