// toEncoded returns a list of all of the serialized versions of these keys,
// plus a stringset of all the encoded root keys that `keys` represents.
func toEncoded(keys []*datastore.Key) (full []string, roots stringset.Set) {
	roots = stringset.New(len(keys))
	full = make([]string, len(keys))
	for i, k := range keys {
		roots.Add(string(serialize.ToBytes(k.Root())))
		full[i] = string(serialize.ToBytes(k))
	}
	return
}

func TestDatastoreQueries(t *testing.T) {
	Convey("Datastore Query suport", t, func() {
		c := Use(context.Background())
		ds := dsS.Get(c)
		So(ds, ShouldNotBeNil)

		Convey("can create good queries", func() {
			q := ds.NewQuery("Foo").Filter("farnsworth >", 20).KeysOnly().Limit(10).Offset(39)

			// normally you can only get cursors from inside of the memory
			// implementation, so this construction is just for testing.
			start := queryCursor(bjoin(
				mkNum(2),
				serialize.ToBytes(dsS.IndexColumn{Property: "farnsworth"}),
				serialize.ToBytes(dsS.IndexColumn{Property: "__key__"}),
				serialize.ToBytes(prop(200)),
				serialize.ToBytes(prop(ds.NewKey("Foo", "id", 0, nil)))))

			So(start.String(), ShouldEqual,
				`gYAAZzFdTeeb3d9zOxsAAF-v221Xy32_AIGHyIgAAUc32-AFabMAAA==`)

			end := queryCursor(bjoin(
				mkNum(2),
				serialize.ToBytes(dsS.IndexColumn{Property: "farnsworth"}),
				serialize.ToBytes(dsS.IndexColumn{Property: "__key__"}),
				serialize.ToBytes(prop(3000)),
				serialize.ToBytes(prop(ds.NewKey("Foo", "zeta", 0, nil)))))

			q = q.Start(start).End(end)
			So(q, ShouldNotBeNil)
			So(q.(*queryImpl).err, ShouldBeNil)
			rq, err := q.(*queryImpl).reduce("", false)
			So(rq, ShouldNotBeNil)
			So(err, ShouldBeNil)
		})

		Convey("ensures orders make sense", func() {
			q := ds.NewQuery("Cool")
			q = q.Filter("cat =", 19).Filter("bob =", 10).Order("bob").Order("bob")

			Convey("removes dups and equality orders", func() {
				q = q.Order("wat")
				qi := q.(*queryImpl)
				So(qi.err, ShouldBeNil)
				rq, err := qi.reduce("", false)
				So(err, ShouldBeNil)
				So(rq.suffixFormat, ShouldResemble, []dsS.IndexColumn{
					{Property: "wat"}, {Property: "__key__"}})
			})

			Convey("if we equality-filter on __key__, that's just silly", func() {
				q = q.Order("wat").Filter("__key__ =", ds.NewKey("Foo", "wat", 0, nil))
				_, err := q.(*queryImpl).reduce("", false)
				So(err, ShouldErrLike, "query equality filter on __key__ is silly")
			})

		})

	})
}

func addIndex(store *memStore, ns string, compIdx []*ds.IndexDefinition) {
	normalized := make([]*ds.IndexDefinition, len(compIdx))
	idxColl := store.SetCollection("idx", nil)
	for i, idx := range compIdx {
		normalized[i] = idx.Normalize()
		idxColl.Set(serialize.ToBytes(*normalized[i].PrepForIdxTable()), []byte{})
	}

	if allEnts := store.GetCollection("ents:" + ns); allEnts != nil {
		allEnts.VisitItemsAscend(nil, true, func(i *gkvlite.Item) bool {
			pm, err := rpmWoCtx(i.Val, ns)
			memoryCorruption(err)

			prop, err := serialize.ReadProperty(bytes.NewBuffer(i.Key), serialize.WithoutContext, globalAppID, ns)
			memoryCorruption(err)

			k := prop.Value().(ds.Key)

			sip := partiallySerialize(k, pm)

			mergeIndexes(ns, store,
				newMemStore(),
				sip.indexEntries(ns, normalized))
			return true
		})
	}
}

// walkCompIdxs walks the table of compound indexes in the store. If `endsWith`
// is provided, this will only walk over compound indexes which match
// Kind, Ancestor, and whose SortBy has `endsWith.SortBy` as a suffix.
func walkCompIdxs(store *memStore, endsWith *ds.IndexDefinition, cb func(*ds.IndexDefinition) bool) {
	idxColl := store.GetCollection("idx")
	if idxColl == nil {
		return
	}
	itrDef := iterDefinition{c: idxColl}

	if endsWith != nil {
		full := serialize.ToBytes(*endsWith.Flip())
		// chop off the null terminating byte
		itrDef.prefix = full[:len(full)-1]
	}

	it := itrDef.mkIter()
	defer it.stop()
	for !it.stopped {
		it.next(nil, func(i *gkvlite.Item) {
			if i == nil {
				return
			}
			qi, err := serialize.ReadIndexDefinition(bytes.NewBuffer(i.Key))
			memoryCorruption(err)
			if !cb(qi.Flip()) {
				it.stop()
			}
		})
	}
}

func (d *dataStoreData) putMulti(keys []ds.Key, vals []ds.PropertyMap, cb ds.PutMultiCB) {
	for i, k := range keys {
		pmap, _ := vals[i].Save(false)
		dataBytes := serialize.ToBytes(pmap)

		k, err := func() (ret ds.Key, err error) {
			d.rwlock.Lock()
			defer d.rwlock.Unlock()

			ents, ret := d.entsKeyLocked(k)
			incrementLocked(ents, groupMetaKey(ret))

			old := ents.Get(keyBytes(ret))
			oldPM := ds.PropertyMap(nil)
			if old != nil {
				if oldPM, err = rpmWoCtx(old, ret.Namespace()); err != nil {
					return
				}
			}
			updateIndexes(d.head, ret, oldPM, pmap)
			ents.Set(keyBytes(ret), dataBytes)
			return
		}()
		if cb != nil {
			cb(k, err)
		}
	}
}

func incrementLocked(ents *memCollection, key []byte) int64 {
	ret := curVersion(ents, key) + 1
	ents.Set(key, serialize.ToBytes(ds.PropertyMap{
		"__version__": {ds.MkPropertyNI(ret)},
	}))
	return ret
}

func HashKey(k datastore.Key) string {
	dgst := sha1.Sum(serialize.ToBytes(k))
	buf := bytes.Buffer{}
	enc := base64.NewEncoder(base64.StdEncoding, &buf)
	_, _ = enc.Write(dgst[:])
	enc.Close()
	return buf.String()[:buf.Len()-Sha1B64Padding]
}

func (q *queryImpl) addEqFilt(prop string, p ds.Property) {
	binVal := string(serialize.ToBytes(p))
	if cur, ok := q.eqFilters[prop]; !ok {
		q.eqFilters[prop] = stringSet{binVal: {}}
	} else {
		cur.add(binVal)
	}
}

// GetBinaryBounds gets the binary encoding of the upper and lower bounds of
// the inequality filter on fq, if any is defined. If a bound does not exist,
// it is nil.
//
// NOTE: if fq specifies a descending sort order for the inequality, the bounds
// will be inverted, incremented, and flipped.
func GetBinaryBounds(fq *ds.FinalizedQuery) (lower, upper []byte) {
	// Pick up the start/end range from the inequalities, if any.
	//
	// start and end in the reducedQuery are normalized so that `start >=
	// X < end`. Because of that, we need to tweak the inequality filters
	// contained in the query if they use the > or <= operators.
	if ineqProp := fq.IneqFilterProp(); ineqProp != "" {
		_, startOp, startV := fq.IneqFilterLow()
		if startOp != "" {
			lower = serialize.ToBytes(startV)
			if startOp == ">" {
				lower = increment(lower)
			}
		}

		_, endOp, endV := fq.IneqFilterHigh()
		if endOp != "" {
			upper = serialize.ToBytes(endV)
			if endOp == "<=" {
				upper = increment(upper)
			}
		}

		// The inequality is specified in natural (ascending) order in the query's
		// Filter syntax, but the order information may indicate to use a descending
		// index column for it. If that's the case, then we must invert, swap and
		// increment the inequality endpoints.
		//
		// Invert so that the desired numbers are represented correctly in the index.
		// Swap so that our iterators still go from >= start to < end.
		// Increment so that >= and < get correctly bounded (since the iterator is
		// still using natrual bytes ordering)
		if fq.Orders()[0].Descending {
			hi, lo := []byte(nil), []byte(nil)
			if len(lower) > 0 {
				lo = increment(serialize.Invert(lower))
			}
			if len(upper) > 0 {
				hi = increment(serialize.Invert(upper))
			}
			upper, lower = lo, hi
		}
	}
	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
}

func incrementLocked(ents *memCollection, key []byte, amt int) int64 {
	if amt <= 0 {
		panic(fmt.Errorf("incrementLocked called with bad `amt`: %d", amt))
	}
	ret := curVersion(ents, key) + 1
	ents.Set(key, serialize.ToBytes(ds.PropertyMap{
		"__version__": {ds.MkPropertyNI(ret + int64(amt-1))},
	}))
	return ret
}

func TestCompoundIndexes(t *testing.T) {
	t.Parallel()

	idxKey := func(def dsS.IndexDefinition) string {
		So(def, ShouldNotBeNil)
		return "idx::" + string(serialize.ToBytes(*def.PrepForIdxTable()))
	}

	numItms := func(c *memCollection) uint64 {
		ret, _ := c.GetTotals()
		return ret
	}

	Convey("Test Compound indexes", t, func() {
		type Model struct {
			ID int64 `gae:"$id"`

			Field1 []string
			Field2 []int64
		}

		c := Use(context.Background())
		ds := dsS.Get(c)
		t := ds.Testable().(*dsImpl)
		head := t.data.head

		So(ds.Put(&Model{1, []string{"hello", "world"}, []int64{10, 11}}), ShouldBeNil)

		idx := dsS.IndexDefinition{
			Kind: "Model",
			SortBy: []dsS.IndexColumn{
				{Property: "Field2"},
			},
		}

		coll := head.GetCollection(idxKey(idx))
		So(coll, ShouldNotBeNil)
		So(numItms(coll), ShouldEqual, 2)

		idx.SortBy[0].Property = "Field1"
		coll = head.GetCollection(idxKey(idx))
		So(coll, ShouldNotBeNil)
		So(numItms(coll), ShouldEqual, 2)

		idx.SortBy = append(idx.SortBy, dsS.IndexColumn{Property: "Field1"})
		So(head.GetCollection(idxKey(idx)), ShouldBeNil)

		t.AddIndexes(&idx)
		coll = head.GetCollection(idxKey(idx))
		So(coll, ShouldNotBeNil)
		So(numItms(coll), ShouldEqual, 4)
	})
}

func serializeRow(vals []ds.Property) serializedPvals {
	dups := map[string]struct{}{}
	ret := make(serializedPvals, 0, len(vals))
	for _, v := range vals {
		if v.IndexSetting() == ds.NoIndex {
			continue
		}
		data := serialize.ToBytes(v.ForIndex())
		dataS := string(data)
		if _, ok := dups[dataS]; ok {
			continue
		}
		dups[dataS] = struct{}{}
		ret = append(ret, data)
	}
	return ret
}

func (sip serializedIndexablePmap) indexEntries(ns string, idxs []*ds.IndexDefinition) *memStore {
	ret := newMemStore()
	idxColl := ret.SetCollection("idx", nil)

	mtch := matcher{}
	for _, idx := range idxs {
		idx = idx.Normalize()
		if irg, ok := mtch.match(idx.GetFullSortOrder(), sip); ok {
			idxBin := serialize.ToBytes(*idx.PrepForIdxTable())
			idxColl.Set(idxBin, []byte{})
			coll := ret.SetCollection(fmt.Sprintf("idx:%s:%s", ns, idxBin), nil)
			irg.permute(coll.Set)
		}
	}

	return ret
}

func (d *dataStoreData) putMulti(keys []*ds.Key, vals []ds.PropertyMap, cb ds.PutMultiCB) error {
	ns := keys[0].Namespace()

	for i, k := range keys {
		pmap, _ := vals[i].Save(false)
		dataBytes := serialize.ToBytes(pmap)

		k, err := func() (ret *ds.Key, err error) {
			d.Lock()
			defer d.Unlock()

			ents := d.mutableEntsLocked(ns)

			ret, err = d.fixKeyLocked(ents, k)
			if err != nil {
				return
			}
			if !d.disableSpecialEntities {
				incrementLocked(ents, groupMetaKey(ret), 1)
			}

			old := ents.Get(keyBytes(ret))
			oldPM := ds.PropertyMap(nil)
			if old != nil {
				if oldPM, err = rpm(old); err != nil {
					return
				}
			}
			ents.Set(keyBytes(ret), dataBytes)
			updateIndexes(d.head, ret, oldPM, pmap)
			return
		}()
		if cb != nil {
			if err := cb(k, err); err != nil {
				if err == ds.Stop {
					return nil
				}
				return err
			}
		}
	}
	return nil
}

func (i *item) getEncKey() string {
	if i.encKey == "" {
		i.encKey = string(serialize.ToBytes(i.key))
	}
	return i.encKey
}

// maybeAddDefinition possibly adds a new IndexDefinitionSortable to this slice.
// It's only added if it could be useful in servicing q, otherwise this function
// is a noop.
//
// This returns true iff the proposed index is OK and depletes missingTerms to
// empty.
//
// If the proposed index is PERFECT (e.g. contains enough columns to cover all
// equality filters, and also has the correct suffix), idxs will be replaced
// with JUST that index, and this will return true.
func (idxs *IndexDefinitionSortableSlice) maybeAddDefinition(q *reducedQuery, s *memStore, missingTerms stringSet, id *ds.IndexDefinition) bool {
	// Kindless queries are handled elsewhere.
	if id.Kind != q.kind {
		impossible(
			fmt.Errorf("maybeAddDefinition given index with wrong kind %q v %q", id.Kind, q.kind))
	}

	// If we're an ancestor query, and the index is compound, but doesn't include
	// an Ancestor field, it doesn't work. Builtin indexes can be used for
	// ancestor queries (and have !Ancestor), assuming that it's only equality
	// filters (plus inequality on __key__), or a single inequality.
	if q.eqFilters["__ancestor__"] != nil && !id.Ancestor && !id.Builtin() {
		impossible(
			fmt.Errorf("maybeAddDefinition given compound index with wrong ancestor info: %s %#v", id, q))
	}

	// add __ancestor__ if necessary
	sortBy := id.GetFullSortOrder()

	// If the index has fewer fields than we need for the suffix, it can't
	// possibly help.
	if len(sortBy) < len(q.suffixFormat) {
		return false
	}

	numEqFilts := len(sortBy) - len(q.suffixFormat)
	// make sure the orders are precisely the same
	for i, sb := range sortBy[numEqFilts:] {
		if q.suffixFormat[i] != sb {
			return false
		}
	}

	if id.Builtin() && numEqFilts == 0 {
		if len(q.eqFilters) > 1 || (len(q.eqFilters) == 1 && q.eqFilters["__ancestor__"] == nil) {
			return false
		}
	}

	// Make sure the equalities section doesn't contain any properties we don't
	// want in our query.
	//
	// numByProp && totalEqFilts will be used to see if this is a perfect match
	// later.
	numByProp := make(map[string]int, len(q.eqFilters))
	totalEqFilts := 0

	eqFilts := sortBy[:numEqFilts]
	for _, p := range eqFilts {
		if _, ok := q.eqFilters[p.Property]; !ok {
			return false
		}
		numByProp[p.Property]++
		totalEqFilts++
	}

	// ok, we can actually use this

	// Grab the collection for convenience later. We don't want to invalidate this
	// index's potential just because the collection doesn't exist. If it's
	// a builtin and it doesn't exist, it still needs to be one of the 'possible'
	// indexes... it just means that the user's query will end up with no results.
	coll := s.GetCollection(
		fmt.Sprintf("idx:%s:%s", q.ns, serialize.ToBytes(*id.PrepForIdxTable())))

	// First, see if it's a perfect match. If it is, then our search is over.
	//
	// A perfect match contains ALL the equality filter columns (or more, since
	// we can use residuals to fill in the extras).
	toAdd := IndexDefinitionSortable{coll: coll}
	toAdd.eqFilts = eqFilts
	for _, sb := range toAdd.eqFilts {
		missingTerms.rm(sb.Property)
	}

	perfect := false
	if len(sortBy) == q.numCols {
		perfect = true
		for k, num := range numByProp {
			if num < len(q.eqFilters[k]) {
				perfect = false
				break
			}
		}
	}
	if perfect {
		*idxs = IndexDefinitionSortableSlice{toAdd}
	} else {
		*idxs = append(*idxs, toAdd)
	}
//.........这里部分代码省略.........

	{"silly inequality (=> v <=)",
		nq().Gte("bob", 10).Lte("bob", 10),
		nil, nil},

	{"cursors get smooshed into the inquality range",
		(nq().Gt("Foo", 3).Lt("Foo", 10).
			Start(curs("Foo", 2, "__key__", key("Something", 1))).
			End(curs("Foo", 20, "__key__", key("Something", 20)))),
		nil,
		&reducedQuery{
			"dev~app", "ns", "Foo", map[string]stringset.Set{}, []dstore.IndexColumn{
				{Property: "Foo"},
				{Property: "__key__"},
			},
			increment(serialize.ToBytes(dstore.MkProperty(3))),
			serialize.ToBytes(dstore.MkProperty(10)),
			2,
		}},

	{"cursors could cause the whole query to be useless",
		(nq().Gt("Foo", 3).Lt("Foo", 10).
			Start(curs("Foo", 200, "__key__", key("Something", 1))).
			End(curs("Foo", 1, "__key__", key("Something", 20)))),
		dstore.ErrNullQuery,
		nil},
}

func TestQueries(t *testing.T) {
	t.Parallel()

func (q *queryImpl) Filter(fStr string, val interface{}) ds.Query {
	prop := ""
	op := qInvalid
	p := ds.Property{}
	return q.checkMutateClone(
		func() error {
			var err error
			prop, op, err = parseFilter(fStr)
			if err != nil {
				return err
			}

			if q.kind == "" && prop != "__key__" {
				// https://cloud.google.com/appengine/docs/go/datastore/queries#Go_Kindless_queries
				return fmt.Errorf(
					"kindless queries can only filter on __key__, got %q", fStr)
			}

			err = p.SetValue(val, ds.ShouldIndex)
			if err != nil {
				return err
			}

			if p.Type() == ds.PTKey {
				if !p.Value().(ds.Key).Valid(false, globalAppID, q.ns) {
					return ds.ErrInvalidKey
				}
			}

			if prop == "__key__" {
				if op == qEqual {
					return fmt.Errorf(
						"query equality filter on __key__ is silly: %q", fStr)
				}
				if p.Type() != ds.PTKey {
					return fmt.Errorf("__key__ filter value is not a key: %T", val)
				}
			} else if strings.HasPrefix(prop, "__") && strings.HasSuffix(prop, "__") {
				return fmt.Errorf("filter on reserved property: %q", prop)
			}

			if op != qEqual {
				if q.ineqFilter.prop != "" && q.ineqFilter.prop != prop {
					return fmt.Errorf(
						"inequality filters on multiple properties: %q and %q",
						q.ineqFilter.prop, prop)
				}
				if len(q.order) > 0 && q.order[0].Property != prop {
					return fmt.Errorf(
						"first sort order must match inequality filter: %q v %q",
						q.order[0].Property, prop)
				}
			} else {
				for _, p := range q.project {
					if p == prop {
						return fmt.Errorf(
							"cannot project on field which is used in an equality filter: %q",
							prop)
					}
				}
			}
			return err
		},
		func(q *queryImpl) {
			if op == qEqual {
				// add it to eq filters
				q.addEqFilt(prop, p)

				// remove it from sort orders.
				// https://cloud.google.com/appengine/docs/go/datastore/queries#sort_orders_are_ignored_on_properties_with_equality_filters
				toRm := -1
				for i, o := range q.order {
					if o.Property == prop {
						toRm = i
						break
					}
				}
				if toRm >= 0 {
					q.order = append(q.order[:toRm], q.order[toRm+1:]...)
				}
			} else {
				q.ineqFilter.prop = prop
				if q.ineqFilter.constrain(op, serialize.ToBytes(p)) {
					q.err = errQueryDone
				}
			}
		})
}

func keyBytes(key *ds.Key) []byte {
	return serialize.ToBytes(ds.MkProperty(key))
}