func (s *bucketstore) copyRemainingColls(bsf *bucketstorefile,
	collRest []string, compactStore *gkvlite.Store, writeEvery int) error {
	currSnapshot := bsf.store.Snapshot()
	if currSnapshot == nil {
		return fmt.Errorf("compact source snapshot failed: %v", bsf.path)
	}
	defer currSnapshot.Close()
	for _, collName := range collRest {
		collCurr := currSnapshot.GetCollection(collName)
		if collCurr == nil {
			return fmt.Errorf("compact rest coll missing: %v, collName: %v",
				bsf.path, collName)
		}
		collNext := compactStore.SetCollection(collName, nil)
		if collNext == nil {
			return fmt.Errorf("compact rest dest missing: %v, collName: %v",
				bsf.path, collName)
		}
		_, _, err := copyColl(collCurr, collNext, writeEvery)
		if err != nil {
			return err
		}
	}
	return nil
}

func (s *bucketstore) copyVBucketColls(bsf *bucketstorefile,
	collName string, compactStore *gkvlite.Store, writeEvery int) (
	uint16, *gkvlite.Item, error) {
	vbidStr := collName[0 : len(collName)-len(COLL_SUFFIX_CHANGES)]
	vbid, err := strconv.Atoi(vbidStr)
	if err != nil {
		return 0, nil, err
	}
	if vbid < 0 || vbid > MAX_VBID {
		return 0, nil, fmt.Errorf("compact vbid out of range: %v, vbid: %v",
			bsf.path, vbid)
	}
	cName := fmt.Sprintf("%v%s", vbid, COLL_SUFFIX_CHANGES)
	kName := fmt.Sprintf("%v%s", vbid, COLL_SUFFIX_KEYS)
	cDest := compactStore.SetCollection(cName, nil)
	kDest := compactStore.SetCollection(kName, nil)
	if cDest == nil || kDest == nil {
		return 0, nil, fmt.Errorf("compact could not create colls for vbid: %v",
			vbid)
	}
	cCurr := s.coll(cName) // The c prefix in cFooBar means 'changes'.
	kCurr := s.coll(kName) // The k prefix in kFooBar means 'keys'.
	if cCurr == nil || kCurr == nil {
		return 0, nil, fmt.Errorf("compact source colls missing: %v, vbid: %v",
			bsf.path, vbid)
	}
	// Get a consistent snapshot (keys reflect all changes) of the
	// keys & changes collections.
	ps := s.partitions[uint16(vbid)]
	if ps == nil {
		return 0, nil, fmt.Errorf("compact missing partition for vbid: %v", vbid)
	}
	var currSnapshot *gkvlite.Store
	ps.mutate(func(key, changes *gkvlite.Collection) {
		currSnapshot = bsf.store.Snapshot()
	})
	if currSnapshot == nil {
		return 0, nil, fmt.Errorf("compact source snapshot failed: %v, vbid: %v",
			bsf.path, vbid)
	}
	cCurrSnapshot := currSnapshot.GetCollection(cName)
	kCurrSnapshot := currSnapshot.GetCollection(kName)
	if cCurrSnapshot == nil || kCurrSnapshot == nil {
		return 0, nil, fmt.Errorf("compact missing colls from snapshot: %v, vbid: %v",
			bsf.path, vbid)
	}
	// TODO: Record stats on # changes processed.
	_, lastChange, err := copyColl(cCurrSnapshot, cDest, writeEvery)
	if err != nil {
		return 0, nil, err
	}
	// TODO: Record stats on # keys processed.
	_, _, err = copyColl(kCurrSnapshot, kDest, writeEvery)
	if err != nil {
		return 0, nil, err
	}
	return uint16(vbid), lastChange, err
}

func copyDelta(lastChangeCAS []byte, cName string, kName string,
	srcStore *gkvlite.Store, dstStore *gkvlite.Store,
	writeEvery int) (numVisits uint64, err error) {
	cSrc := srcStore.GetCollection(cName)
	cDst := dstStore.GetCollection(cName)
	kDst := dstStore.GetCollection(kName)
	if cSrc == nil || cDst == nil || kDst == nil {
		return 0, fmt.Errorf("compact copyDelta missing colls: %v, %v",
			cName, kName)
	}

	var errVisit error

	err = cSrc.VisitItemsAscend(lastChangeCAS, true, func(cItem *gkvlite.Item) bool {
		numVisits++
		if numVisits <= 1 {
			return true
		}
		if errVisit = cDst.SetItem(cItem); errVisit != nil {
			return false
		}
		// Update the keys index with the latest change.
		i := &item{}
		if errVisit = i.fromValueBytes(cItem.Val); errVisit != nil {
			return false
		}
		if i.key == nil || len(i.key) <= 0 {
			return true // A nil/empty key means a metadata change.
		}
		if i.isDeletion() {
			if _, errVisit = kDst.Delete(i.key); errVisit != nil {
				return false
			}
		} else {
			if errVisit = kDst.Set(i.key, cItem.Key); errVisit != nil {
				return false
			}
		}
		// Persist to storage as needed.
		if writeEvery > 0 && numVisits%uint64(writeEvery) == 0 {
			if errVisit = cDst.Write(); errVisit != nil {
				return false
			}
			if errVisit = kDst.Write(); errVisit != nil {
				return false
			}
		}
		return true
	})
	if err != nil {
		return 0, err
	}
	if errVisit != nil {
		return 0, errVisit
	}

	return numVisits, nil
}

func NewPster(dbpath string, errlog *log.Logger) (*SimplePster, error) { // {{{1
	var store *gkvlite.Store
	file, err := os.OpenFile(dbpath, os.O_RDWR|os.O_CREATE|os.O_SYNC, 0660)
	if err != nil {
		return nil, err
	}
	store, err = gkvlite.NewStore(file)
	if err != nil {
		return nil, err
	}
	return &SimplePster{
		file:    file,
		store:   store,
		rlog:    store.SetCollection("rlog", nil),
		rfields: store.SetCollection("rfields", nil),
		err:     errlog,
	}, nil
}