Add transaction buffer filter.

impl/memory/datastore_query.go

Index: impl/memory/datastore_query.go
diff --git a/impl/memory/datastore_query.go b/impl/memory/datastore_query.go
index ed49a79820546cfda2b50916961a24563b302a4d..856f4c19bf0c110f29999217dfff28464245c12c 100644
--- a/impl/memory/datastore_query.go
+++ b/impl/memory/datastore_query.go
@@ -102,57 +102,32 @@ func numComponents(fq *ds.FinalizedQuery) int {
 	return numComponents
 }
 
-func reduce(fq *ds.FinalizedQuery, ns string, isTxn bool) (*reducedQuery, error) {
-	if err := fq.Valid(globalAppID, ns); err != nil {
-		return nil, err
-	}
-	if isTxn && fq.Ancestor() == nil {
-		return nil, fmt.Errorf("queries within a transaction must include an Ancestor filter")
-	}
-	if num := numComponents(fq); num > MaxQueryComponents {
-		return nil, fmt.Errorf(
-			"gae/memory: query is too large. may not have more than "+
-				"%d filters + sort orders + ancestor total: had %d",
-			MaxQueryComponents, num)
-	}
-
-	ret := &reducedQuery{
-		ns:           ns,
-		kind:         fq.Kind(),
-		suffixFormat: fq.Orders(),
-	}
-
-	eqFilts := fq.EqFilters()
-	ret.eqFilters = make(map[string]stringset.Set, len(eqFilts))
-	for prop, vals := range eqFilts {
-		sVals := stringset.New(len(vals))
-		for _, v := range vals {
-			sVals.Add(string(serialize.ToBytes(v)))
-		}
-		ret.eqFilters[prop] = sVals
-	}
-
+// 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 fliped.
+func GetBinaryBounds(fq *ds.FinalizedQuery) (lower, upper []byte) {

[iannucci, 2015/09/29 04:43:27]

This was factored out to allow toComparableString in txnBuf to use the same
logic.

 	// 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.
-	startD := []byte(nil)
-	endD := []byte(nil)
 	if ineqProp := fq.IneqFilterProp(); ineqProp != "" {
 		_, startOp, startV := fq.IneqFilterLow()
 		if startOp != "" {
-			startD = serialize.ToBytes(startV)
+			lower = serialize.ToBytes(startV)
 			if startOp == ">" {
-				startD = increment(startD)
+				lower = increment(lower)
 			}
 		}
 
 		_, endOp, endV := fq.IneqFilterHigh()
 		if endOp != "" {
-			endD = serialize.ToBytes(endV)
+			upper = serialize.ToBytes(endV)
 			if endOp == "<=" {
-				endD = increment(endD)
+				upper = increment(upper)
 			}
 		}
 
@@ -165,18 +140,53 @@ func reduce(fq *ds.FinalizedQuery, ns string, isTxn bool) (*reducedQuery, error)
 		// 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 ret.suffixFormat[0].Descending {
+		if fq.Orders()[0].Descending {
 			hi, lo := []byte(nil), []byte(nil)
-			if len(startD) > 0 {
-				lo = increment(serialize.Invert(startD))
+			if len(lower) > 0 {
+				lo = increment(serialize.Invert(lower))
 			}
-			if len(endD) > 0 {
-				hi = increment(serialize.Invert(endD))
+			if len(upper) > 0 {
+				hi = increment(serialize.Invert(upper))
 			}
-			endD, startD = lo, hi
+			upper, lower = lo, hi
+		}
+	}
+	return
+}
+
+func reduce(fq *ds.FinalizedQuery, aid, ns string, isTxn bool) (*reducedQuery, error) {
+	if err := fq.Valid(aid, ns); err != nil {
+		return nil, err
+	}
+	if isTxn && fq.Ancestor() == nil {
+		return nil, fmt.Errorf("queries within a transaction must include an Ancestor filter")
+	}
+	if num := numComponents(fq); num > MaxQueryComponents {
+		return nil, fmt.Errorf(
+			"gae/memory: query is too large. may not have more than "+
+				"%d filters + sort orders + ancestor total: had %d",
+			MaxQueryComponents, num)
+	}
+
+	ret := &reducedQuery{
+		aid:          aid,
+		ns:           ns,
+		kind:         fq.Kind(),
+		suffixFormat: fq.Orders(),
+	}
+
+	eqFilts := fq.EqFilters()
+	ret.eqFilters = make(map[string]stringset.Set, len(eqFilts))
+	for prop, vals := range eqFilts {
+		sVals := stringset.New(len(vals))
+		for _, v := range vals {
+			sVals.Add(string(serialize.ToBytes(v)))
 		}
+		ret.eqFilters[prop] = sVals
 	}
 
+	startD, endD := GetBinaryBounds(fq)
+
 	// Now we check the start and end cursors.
 	//
 	// Cursors are composed of a list of IndexColumns at the beginning, followed