func itemValLength(coll *gkvlite.Collection, i *gkvlite.Item) int {
	if !strings.HasSuffix(coll.Name(), COLL_SUFFIX_CHANGES) {
		return len(i.Val)
	}
	if i.Val != nil {
		return len(i.Val)
	}
	if i.Transient == unsafe.Pointer(nil) {
		// TODO: The item might have nil Val when gkvlite is
		// traversing with a withValue of false; so we haven't
		// read/unmarshal'ed the Val/Transient yet.  Impact: the byte
		// aggregate math might wrong, and if we try to write with
		// this 0 length, it'll be wrong; but, assuming here that the
		// item.Loc() is non-empty so we'll never try to write this
		// nil Val/Transient.
		// panic(fmt.Sprintf("itemValLength saw nil Transient, i: %#v, coll.name: %v",
		// 	i, coll.Name()))
		return 0
	}
	ti := (interface{})(i.Transient)
	item, ok := ti.(*item)
	if !ok {
		panic(fmt.Sprintf("itemValLength invoked on non-item, i: %#v", i))
	}
	if item == nil {
		panic(fmt.Sprintf("itemValLength invoked on nil item, i: %#v", i))
	}
	return itemHdrLen + len(item.key) + len(item.data)
}

func itemValWrite(coll *gkvlite.Collection, i *gkvlite.Item,
	w io.WriterAt, offset int64) error {
	if !strings.HasSuffix(coll.Name(), COLL_SUFFIX_CHANGES) {
		_, err := w.WriteAt(i.Val, offset)
		return err
	}
	if i.Val != nil {
		_, err := w.WriteAt(i.Val, offset)
		return err
	}
	if i.Transient == unsafe.Pointer(nil) {
		panic(fmt.Sprintf("itemValWrite saw nil Transient, i: %#v", i))
	}
	ti := (interface{})(i.Transient)
	item, ok := ti.(*item)
	if !ok {
		panic(fmt.Sprintf("itemValWrite invoked on non-item, i: %#v", i))
	}
	if item == nil {
		panic(fmt.Sprintf("itemValWrite invoked on nil item, i: %#v", i))
	}
	vBytes := item.toValueBytes()
	_, err := w.WriteAt(vBytes, offset)
	return err
}

func collRangeCopy(src *gkvlite.Collection, dst *gkvlite.Collection,
	minKey []byte,
	minKeyInclusive []byte,
	maxKeyExclusive []byte) error {
	var errVisit error
	visitor := func(i *gkvlite.Item) bool {
		if len(minKeyInclusive) > 0 &&
			bytes.Compare(i.Key, minKeyInclusive) < 0 {
			return true
		}
		if len(maxKeyExclusive) > 0 &&
			bytes.Compare(i.Key, maxKeyExclusive) >= 0 {
			return true
		}
		errVisit = dst.SetItem(i)
		if errVisit != nil {
			return false
		}
		return true
	}
	if errVisit != nil {
		return errVisit
	}
	return src.VisitItemsAscend(minKey, true, visitor)
}

func (bs *BasicStore) withRecordContext(id string, col *gkvlite.Collection, fn func(record []byte) error) error {
	record, err := col.Get([]byte(id))
	if err != nil {
		return &taurus.StoreError{Code: taurus.ErrFailedRead, Err: err}
	}
	return fn(record)
}

func testFillColl(x *gkvlite.Collection, arr []string) {
	for i, s := range arr {
		x.SetItem(&gkvlite.Item{
			Key:      []byte(s),
			Val:      []byte(s),
			Priority: int32(i),
		})
	}
}

// You can use fmt.Printf() for the printf param.
func dumpColl(printf func(format string, a ...interface{}) (n int, err error),
	c *gkvlite.Collection, prefix string) (int, error) {
	n := 0
	err := c.VisitItemsAscend(nil, true, func(cItem *gkvlite.Item) bool {
		n++
		printf("%v%s %#v\n", prefix, string(cItem.Key), cItem)
		return true
	})
	return n, err
}

func colItemExists(key string, col *gkvlite.Collection) bool {
	exists := false
	col.VisitItemsAscend([]byte(""), true, func(i *gkvlite.Item) bool {
		if key == string(i.Key) {
			exists = true
			return false
		}
		return true
	})
	return exists
}

func copyColl(srcColl *gkvlite.Collection, dstColl *gkvlite.Collection,
	writeEvery int) (numItems uint64, lastItem *gkvlite.Item, err error) {
	minItem, err := srcColl.MinItem(true)
	if err != nil {
		return 0, nil, err
	}
	if minItem == nil {
		return 0, nil, nil
	}

	var errVisit error = nil
	err = srcColl.VisitItemsAscend(minItem.Key, true, func(i *gkvlite.Item) bool {
		if errVisit = dstColl.SetItem(i); errVisit != nil {
			return false
		}
		numItems++
		lastItem = i
		if writeEvery > 0 && numItems%uint64(writeEvery) == 0 {
			if errVisit = dstColl.Write(); errVisit != nil {
				return false
			}
		}
		return true
	})
	if err != nil {
		return 0, nil, err
	}
	if errVisit != nil {
		return 0, nil, errVisit
	}

	return numItems, lastItem, nil
}

func (p *partitionstore) visit(coll *gkvlite.Collection,
	start []byte, withValue bool,
	v func(*gkvlite.Item) bool) (err error) {
	if start == nil {
		i, err := coll.MinItem(false)
		if err != nil {
			return err
		}
		if i == nil {
			return nil
		}
		start = i.Key
	}
	return coll.VisitItemsAscend(start, withValue, v)
}

// You can use fmt.Printf() for the printf param.
func dumpCollAsItems(printf func(format string, a ...interface{}) (n int, err error),
	c *gkvlite.Collection, prefix string) (int, error) {
	n := 0
	var vErr error
	err := c.VisitItemsAscend(nil, true, func(cItem *gkvlite.Item) bool {
		i := &item{}
		if vErr = i.fromValueBytes(cItem.Val); vErr != nil {
			return false
		}
		n++
		printf("%v%#v, data: %v\n", prefix, i, string(i.data))
		return true
	})
	if vErr != nil {
		return 0, vErr
	}
	return n, err
}

func rangeCopy(srcColl *gkvlite.Collection, dstColl *gkvlite.Collection,
	minKeyInclusive []byte, maxKeyExclusive []byte) (hasItems bool, err error) {
	minItem, err := srcColl.MinItem(false)
	if err != nil {
		return false, err
	}
	// TODO: What if we flush between the keys update and changes
	// update?  That could result in an inconsistent db file?
	// Solution idea #1 is to have load-time fixup, that
	// incorporates changes into the key-index.
	if minItem != nil {
		if err := collRangeCopy(srcColl, dstColl, minItem.Key,
			minKeyInclusive, maxKeyExclusive); err != nil {
			return false, err
		}
		return true, nil
	}
	return false, nil
}

func itemValRead(coll *gkvlite.Collection, i *gkvlite.Item,
	r io.ReaderAt, offset int64, valLength uint32) error {
	if i.Val != nil {
		panic(fmt.Sprintf("itemValRead saw non-nil Val, i: %#v", i))
	}
	i.Val = make([]byte, valLength)
	_, err := r.ReadAt(i.Val, offset)
	if err != nil {
		return err
	}
	if !strings.HasSuffix(coll.Name(), COLL_SUFFIX_CHANGES) {
		return nil
	}
	x := &item{}
	if err = x.fromValueBytes(i.Val); err != nil {
		return err
	}
	atomic.StorePointer(&i.Transient, unsafe.Pointer(x))
	return nil
}

func testCheckColl(t *testing.T, x *gkvlite.Collection, start string, arr []string, cb func(i *gkvlite.Item)) {
	n := 0
	err := x.VisitItemsAscend([]byte(start), true, func(i *gkvlite.Item) bool {
		if cb != nil {
			cb(i)
		}
		if n >= len(arr) {
			t.Errorf("visited more than expected: %v, saw: %v", len(arr), n+1)
		}
		if string(i.Key) != arr[n] {
			t.Errorf("expected visit item: %v, saw: %v", arr[n], i)
		}
		n++
		return true
	})
	if err != nil {
		t.Errorf("expected no visit error, got: %v", err)
	}
	if n != len(arr) {
		t.Errorf("expected # visit callbacks: %v, saw: %v", len(arr), n)
	}
}