func (d *dataStoreData) allocateIDs(incomplete *ds.Key, n int) (int64, error) {
	d.Lock()
	defer d.Unlock()

	ents := d.mutableEntsLocked(incomplete.Namespace())
	return d.allocateIDsLocked(ents, incomplete, n)
}

// writeMutation ensures that this transaction can support the given key/value
// mutation.
//
//   if getOnly is true, don't record the actual mutation data, just ensure that
//	   the key is in an included entity group (or add an empty entry for that
//	   group).
//
//   if !getOnly && data == nil, this counts as a deletion instead of a Put.
//
// Returns an error if this key causes the transaction to cross too many entity
// groups.
func (td *txnDataStoreData) writeMutation(getOnly bool, key *ds.Key, data ds.PropertyMap) error {
	rk := string(keyBytes(key.Root()))

	td.Lock()
	defer td.Unlock()

	if _, ok := td.muts[rk]; !ok {
		limit := 1
		if td.isXG {
			limit = xgEGLimit
		}
		if len(td.muts)+1 > limit {
			msg := "cross-group transaction need to be explicitly specified (xg=True)"
			if td.isXG {
				msg = "operating on too many entity groups in a single transaction"
			}
			return errors.New(msg)
		}
		td.muts[rk] = []txnMutation{}
	}
	if !getOnly {
		td.muts[rk] = append(td.muts[rk], txnMutation{key, data})
	}

	return nil
}

// Split splits the key into its constituent parts. Note that if the key is
// not Valid, this method may not provide a round-trip for k.
func Split(k ds.Key) (appID, namespace string, toks []ds.KeyTok) {
	if k == nil {
		return
	}

	if sk, ok := k.(*Generic); ok {
		if sk == nil {
			return
		}
		return sk.appID, sk.namespace, sk.toks
	}

	n := 0
	for i := k; i != nil; i = i.Parent() {
		n++
	}
	toks = make([]ds.KeyTok, n)
	for i := k; i != nil; i = i.Parent() {
		n--
		toks[n].IntID = i.IntID()
		toks[n].StringID = i.StringID()
		toks[n].Kind = i.Kind()
	}
	appID = k.AppID()
	namespace = k.Namespace()
	return
}

func testGetMeta(c context.Context, k *dsS.Key) int64 {
	ds := dsS.Get(c)
	mg := &MetaGroup{Parent: k.Root()}
	if err := ds.Get(mg); err != nil {
		panic(err)
	}
	return mg.Version
}

func (d rdsImpl) AllocateIDs(incomplete *ds.Key, n int) (start int64, err error) {
	par, err := dsF2R(d.aeCtx, incomplete.Parent())
	if err != nil {
		return
	}

	start, _, err = datastore.AllocateIDs(d.aeCtx, incomplete.Kind(), par, n)
	return
}

func (d *dataStoreData) fixKeyLocked(ents *memCollection, key *ds.Key) (*ds.Key, error) {
	if key.Incomplete() {
		id, err := d.allocateIDsLocked(ents, key, 1)
		if err != nil {
			return key, err
		}
		key = ds.NewKey(key.AppID(), key.Namespace(), key.Kind(), "", id, key.Parent())
	}
	return key, nil
}

func updateIndexes(store *memStore, key ds.Key, oldEnt, newEnt ds.PropertyMap) {
	// load all current complex query index definitions.
	compIdx := []*ds.IndexDefinition{}
	walkCompIdxs(store, nil, func(i *ds.IndexDefinition) bool {
		compIdx = append(compIdx, i)
		return true
	})

	mergeIndexes(key.Namespace(), store,
		indexEntriesWithBuiltins(key, oldEnt, compIdx),
		indexEntriesWithBuiltins(key, newEnt, compIdx))
}

func (d *dataStoreData) allocateIDsLocked(ents *memCollection, incomplete *ds.Key, n int) (int64, error) {
	if d.disableSpecialEntities {
		return 0, errors.New("disableSpecialEntities is true so allocateIDs is disabled")
	}

	idKey := []byte(nil)
	if incomplete.Parent() == nil {
		idKey = rootIDsKey(incomplete.Kind())
	} else {
		idKey = groupIDsKey(incomplete)
	}
	return incrementLocked(ents, idKey, n), nil
}

// PropertyMapPartially turns a regular PropertyMap into a SerializedPmap.
// Essentially all the []Property's become SerializedPslice, using cmpbin and
// datastore/serialize's encodings.
func PropertyMapPartially(k *ds.Key, pm ds.PropertyMap) (ret SerializedPmap) {
	ret = make(SerializedPmap, len(pm)+2)
	if k != nil {
		ret["__key__"] = [][]byte{ToBytes(ds.MkProperty(k))}
		for k != nil {
			ret["__ancestor__"] = append(ret["__ancestor__"], ToBytes(ds.MkProperty(k)))
			k = k.Parent()
		}
	}
	for k, vals := range pm {
		newVals := PropertySlice(vals)
		if len(newVals) > 0 {
			ret[k] = newVals
		}
	}
	return
}

func partiallySerialize(k ds.Key, pm ds.PropertyMap) (ret serializedIndexablePmap) {
	ret = make(serializedIndexablePmap, len(pm)+2)
	if k == nil {
		impossible(fmt.Errorf("key to partiallySerialize is nil"))
	}
	ret["__key__"] = [][]byte{serialize.ToBytes(ds.MkProperty(k))}
	for k != nil {
		ret["__ancestor__"] = append(ret["__ancestor__"], serialize.ToBytes(ds.MkProperty(k)))
		k = k.Parent()
	}
	for k, vals := range pm {
		newVals := serializeRow(vals)
		if len(newVals) > 0 {
			ret[k] = newVals
		}
	}
	return
}

// WriteKey encodes a key to the buffer. If context is WithContext, then this
// encoded value will include the appid and namespace of the key.
func WriteKey(buf Buffer, context KeyContext, k *ds.Key) (err error) {
	// [appid ++ namespace]? ++ [1 ++ token]* ++ NULL
	defer recoverTo(&err)
	appid, namespace, toks := k.Split()
	if context == WithContext {
		panicIf(buf.WriteByte(1))
		_, e := cmpbin.WriteString(buf, appid)
		panicIf(e)
		_, e = cmpbin.WriteString(buf, namespace)
		panicIf(e)
	} else {
		panicIf(buf.WriteByte(0))
	}
	for _, tok := range toks {
		panicIf(buf.WriteByte(1))
		panicIf(WriteKeyTok(buf, tok))
	}
	return buf.WriteByte(0)
}

// dsF2R (DS fake-to-real) converts a DSKey back to an SDK *Key.
func dsF2R(aeCtx context.Context, k *ds.Key) (*datastore.Key, error) {
	if k == nil {
		return nil, nil
	}

	// drop aid.
	_, ns, toks := k.Split()
	err := error(nil)
	aeCtx, err = appengine.Namespace(aeCtx, ns)
	if err != nil {
		return nil, err
	}

	ret := datastore.NewKey(aeCtx, toks[0].Kind, toks[0].StringID, toks[0].IntID, nil)
	for _, t := range toks[1:] {
		ret = datastore.NewKey(aeCtx, t.Kind, t.StringID, t.IntID, ret)
	}

	return ret, nil
}

func (q *queryImpl) Ancestor(k ds.Key) ds.Query {
	return q.checkMutateClone(
		func() error {
			if k == nil {
				// SDK has an explicit nil-check
				return errors.New("datastore: nil query ancestor")
			}
			if k.Namespace() != q.ns {
				return fmt.Errorf("bad namespace: %q (expected %q)", k.Namespace(), q.ns)
			}
			if !k.Valid(false, globalAppID, q.ns) {
				// technically the SDK implementation does a Weird Thing (tm) if both the
				// stringID and intID are set on a key; it only serializes the stringID in
				// the proto. This means that if you set the Ancestor to an invalid key,
				// you'll never actually hear about it. Instead of doing that insanity, we
				// just swap to an error here.
				return ds.ErrInvalidKey
			}
			if q.eqFilters["__ancestor__"] != nil {
				return errors.New("cannot have more than one ancestor")
			}
			return nil
		},
		func(q *queryImpl) {
			q.addEqFilt("__ancestor__", ds.MkProperty(k))
		})
}

// Encode encodes the provided key as a base64-encoded protobuf.
//
// This encoding is compatible with the SDK-provided encoding and is agnostic
// to the underlying implementation of the Key.
//
// It's encoded with the urlsafe base64 table.
func Encode(k ds.Key) string {
	n := 0
	for i := k; i != nil; i = i.Parent() {
		n++
	}
	e := make([]*pb.Path_Element, n)
	for i := k; i != nil; i = i.Parent() {
		n--
		kind := i.Kind()
		e[n] = &pb.Path_Element{
			Type: &kind,
		}
		// At most one of {Name,Id} should be set.
		// Neither will be set for incomplete keys.
		if i.StringID() != "" {
			sid := i.StringID()
			e[n].Name = &sid
		} else if i.IntID() != 0 {
			iid := i.IntID()
			e[n].Id = &iid
		}
	}
	var namespace *string
	if k.Namespace() != "" {
		namespace = proto.String(k.Namespace())
	}
	r, err := proto.Marshal(&pb.Reference{
		App:       proto.String(k.AppID()),
		NameSpace: namespace,
		Path: &pb.Path{
			Element: e,
		},
	})
	if err != nil {
		panic(err)
	}

	// trim padding
	return strings.TrimRight(base64.URLEncoding.EncodeToString(r), "=")
}

// Incomplete returns true iff k doesn't have an id yet.
func Incomplete(k ds.Key) bool {
	return k != nil && k.StringID() == "" && k.IntID() == 0
}

// Valid determines if a key is valid, according to a couple rules:
//   - k is not nil
//   - every token of k:
//     - (if !allowSpecial) token's kind doesn't start with '__'
//     - token's kind and appid are non-blank
//     - token is not incomplete
//   - all tokens have the same namespace and appid
func Valid(k ds.Key, allowSpecial bool, aid, ns string) bool {
	if k == nil {
		return false
	}
	if aid != k.AppID() || ns != k.Namespace() {
		return false
	}
	for ; k != nil; k = k.Parent() {
		if !allowSpecial && len(k.Kind()) >= 2 && k.Kind()[:2] == "__" {
			return false
		}
		if k.Kind() == "" || k.AppID() == "" {
			return false
		}
		if k.StringID() != "" && k.IntID() != 0 {
			return false
		}
		if k.Parent() != nil {
			if k.Parent().Incomplete() {
				return false
			}
			if k.Parent().AppID() != k.AppID() || k.Parent().Namespace() != k.Namespace() {
				return false
			}
		}
	}
	return true
}

// PartialValid returns true iff this key is suitable for use in a Put
// operation. This is the same as Valid(k, false, ...), but also allowing k to
// be Incomplete().
func PartialValid(k ds.Key, aid, ns string) bool {
	if k.Incomplete() {
		k = New(k.AppID(), k.Namespace(), k.Kind(), "", 1, k.Parent())
	}
	return Valid(k, false, aid, ns)
}

func groupIDsKey(key *ds.Key) []byte {
	return keyBytes(ds.NewKey("", "", "__entity_group_ids__", "", 1, key.Root()))
}

// Root returns the entity root for the given key.
func Root(k ds.Key) ds.Key {
	for k != nil && k.Parent() != nil {
		k = k.Parent()
	}
	return k
}

func indexEntriesWithBuiltins(k ds.Key, pm ds.PropertyMap, complexIdxs []*ds.IndexDefinition) *memStore {
	sip := partiallySerialize(k, pm)
	return sip.indexEntries(k.Namespace(), append(defaultIndexes(k.Kind(), pm), complexIdxs...))
}