Add transaction buffer filter.

impl/memory/datastore_query_execution.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package memory
 
 import (
         "bytes"
         "fmt"
 
@@ skipping 78 matching lines @@
 func (s *keysOnlyStrategy) handle(rawData [][]byte, _ []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) bool {
         if !s.dedup.Add(string(rawData[len(rawData)-1])) {
                 return true
         }
         return s.cb(key, nil, gc)
 }
 
 type normalStrategy struct {
         cb ds.RawRunCB
 
+        aid   string
         ns    string
         head  *memCollection
         dedup stringset.Set
 }
 
-func newNormalStrategy(ns string, cb ds.RawRunCB, head *memStore) queryStrategy {
+func newNormalStrategy(aid, ns string, cb ds.RawRunCB, head *memStore) queryStrategy {
         coll := head.GetCollection("ents:" + ns)
         if coll == nil {
                 return nil
         }
-        return &normalStrategy{cb, ns, coll, stringset.New(0)}
+        return &normalStrategy{cb, aid, ns, coll, stringset.New(0)}
 }
 
 func (s *normalStrategy) handle(rawData [][]byte, _ []ds.Property, key *ds.Key, gc func() (ds.Cursor, error)) bool {
         rawKey := rawData[len(rawData)-1]
         if !s.dedup.Add(string(rawKey)) {
                 return true
         }
 
         rawEnt := s.head.Get(rawKey)
         if rawEnt == nil {
                 // entity doesn't exist at head
                 return true
         }
-        pm, err := serialize.ReadPropertyMap(bytes.NewBuffer(rawEnt), serialize.WithoutContext, globalAppID, s.ns)
+        pm, err := serialize.ReadPropertyMap(bytes.NewBuffer(rawEnt), serialize.WithoutContext, s.aid, s.ns)
         memoryCorruption(err)
 
         return s.cb(key, pm, gc)
 }
 
 func pickQueryStrategy(fq *ds.FinalizedQuery, rq *reducedQuery, cb ds.RawRunCB, head *memStore) queryStrategy {
         if fq.KeysOnly() {
                 return &keysOnlyStrategy{cb, stringset.New(0)}
         }
         if len(fq.Project()) > 0 {
                 return newProjectionStrategy(fq, rq, cb)
         }
-        return newNormalStrategy(rq.ns, cb, head)
+        return newNormalStrategy(rq.aid, rq.ns, cb, head)
 }
 
-func parseSuffix(ns string, suffixFormat []ds.IndexColumn, suffix []byte, count int) (raw [][]byte, decoded []ds.Property) {
+func parseSuffix(aid, ns string, suffixFormat []ds.IndexColumn, suffix []byte, count int) (raw [][]byte, decoded []ds.Property) {
         buf := serialize.Invertible(bytes.NewBuffer(suffix))
         decoded = make([]ds.Property, len(suffixFormat))
         raw = make([][]byte, len(suffixFormat))
 
         err := error(nil)
         for i := range decoded {
-                if count > 0 && i > count {
+                if count >= 0 && i >= count {

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

This was another bug, added a test for it.

                         break
                 }
                 needInvert := suffixFormat[i].Descending
 
                 buf.SetInvert(needInvert)
-                decoded[i], err = serialize.ReadProperty(buf, serialize.WithoutContext, globalAppID, ns)
+                decoded[i], err = serialize.ReadProperty(buf, serialize.WithoutContext, aid, ns)
                 memoryCorruption(err)
 
                 offset := len(suffix) - buf.Len()
                 raw[i] = suffix[:offset]
                 suffix = suffix[offset:]
                 if needInvert {
                         raw[i] = serialize.Invert(raw[i])
                 }
         }
 
         return
 }
 
-func countQuery(fq *ds.FinalizedQuery, ns string, isTxn bool, idx, head *memStore) (ret int64, err error) {
+func countQuery(fq *ds.FinalizedQuery, aid, ns string, isTxn bool, idx, head *memStore) (ret int64, err error) {
         if len(fq.Project()) == 0 && !fq.KeysOnly() {
                 fq, err = fq.Original().KeysOnly(true).Finalize()
                 if err != nil {
                         return
                 }
         }
-        err = executeQuery(fq, ns, isTxn, idx, head, func(_ *ds.Key, _ ds.PropertyMap, _ ds.CursorCB) bool {
+        err = executeQuery(fq, aid, ns, isTxn, idx, head, func(_ *ds.Key, _ ds.PropertyMap, _ ds.CursorCB) bool {
                 ret++
                 return true
         })
         return
 }
 
-func executeQuery(fq *ds.FinalizedQuery, ns string, isTxn bool, idx, head *memStore, cb ds.RawRunCB) error {
-        rq, err := reduce(fq, ns, isTxn)
+func executeQuery(fq *ds.FinalizedQuery, aid, ns string, isTxn bool, idx, head *memStore, cb ds.RawRunCB) error {
+        rq, err := reduce(fq, aid, ns, isTxn)
         if err == ds.ErrNullQuery {
                 return nil
         }
         if err != nil {
                 return err
         }
 
         idxs, err := getIndexes(rq, idx)
         if err == ds.ErrNullQuery {
                 return nil
@@ skipping 37 matching lines @@
                         offset--
                         return true
                 }
                 if hasLimit {
                         if limit <= 0 {
                                 return false
                         }
                         limit--
                 }
 
-                rawData, decodedProps := parseSuffix(ns, rq.suffixFormat, suffix, -1)
+                rawData, decodedProps := parseSuffix(aid, ns, rq.suffixFormat, suffix, -1)
 
                 keyProp := decodedProps[len(decodedProps)-1]
                 if keyProp.Type() != ds.PTKey {
                         impossible(fmt.Errorf("decoded index row doesn't end with a Key: %#v", keyProp))
                 }
 
                 return strategy.handle(
                         rawData, decodedProps, keyProp.Value().(*ds.Key),
                         getCursorFn(suffix))
         })
 
         return nil
 }