impl/memory: Implement Queries

impl/memory/gkvlite_iter.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"
         "sync"
 
         "github.com/luci/gkvlite"
 )
 
-// TODO(riannucci): Add a multi-iterator which allows:
-//   multiple collections
-//     iterate smallest collection -> largest collection
-//     retain input order
-//   constant prefixes for each collection
-//   start + end partial suffixes
-//     * these are appended to the prefix for the collection to narrow the scan
-//       range. The collection scans will start at >= prefix+start suffix, and
-//       will scan until < prefix+end suffix.
-//   produces hits (a suffix+value) when all collections contain the same prefix
-//     and the same suffix
-//     * a hit value will have
-//       - a []byte suffix which is the matching suffix of every collection.
-//         The caller already knows the prefix for each collection. The caller
-//         may need to parse the suffix to separate the sort order(s) from the
-//         Key.
+type iterDefinition struct {

[dnj (Google), 2015/08/23 06:50:07]

Nice combination.

+        // The collection to iterate over
+        c *memCollection
 
-// TODO(riannucci): dedup and distinct will be postfilters
-// dedup could be done by simply analyzing the retrieved entity and then seeing
-//   if it could come before the query parameters, but this will break in
-//   eventually-consistent scenarios. Say that you changed an entity from having
-//   a multiproperty with 100 values to only having 1. Then it shows up in the
-//   index in 100 places, but on retrieval will only have 1 entry (so "it could
-//   not possibly be a dup"). Then you'll serve the entity 100 times :(
-//
-//   Similarly, it won't allow you to dedup on 'distinct' queries, since you'd
-//   be forced to retrieve and decode the entity, defeating the purpose.
+        // The prefix to always assert for every row. A nil prefix matches every row.
+        prefix []byte
 
-func bjoin(itms ...[]byte) []byte {
-        total := 0
-        for _, i := range itms {
-                total += len(i)
-        }
-        ret := make([]byte, 0, total)
-        for _, i := range itms {
-                ret = append(ret, i...)
-        }
-        return ret
-}
+        // prefixLen is the number of prefix bytes that the caller cares about. It
+        // may be <= len(prefix). When doing a multiIterator, this number will be used
+        // to determine the amount of suffix to transfer accross iterators. This is
+        // used specifically when using builtin indexes to service ancestor queries.
+        // The builtin index represents the ancestor key with prefix bytes, but in a
+        // multiIterator context, it wants the entire key to be included in the
+        // suffix.
+        prefixLen int
 
-type iterDefinition struct {
-        c      *memCollection
-        prefix []byte
-        start  []byte
-        end    []byte
+        // The start cursor. It's appended to prefix to find the first row.
+        start []byte
+
+        // The end cursor. It's appended to prefix to find the last row (which is not
+        // included in the interation result). If this is nil, then there's no end
+        // except the natural end of the collection.
+        end []byte
 }
 
 func multiIterate(defs []*iterDefinition, cb func(suffix []byte) bool) {
+        if len(defs) == 0 {
+                return
+        }
+
         ts := make([]*iterator, len(defs))
         prefixLens := make([]int, len(defs))
         for i, def := range defs {
-                if def.prefix == nil {
-                        panic("got nil prefix")
-                }
-
-                end := []byte(nil)
-                if def.end != nil {
-                        end = bjoin(def.prefix, def.end)
-                } else {
-                        end = []byte(increment(string(def.prefix), true))
-                }
-
-                start := def.prefix
-                if def.start != nil {
-                        start = bjoin(def.prefix, def.start)
-                }
-
-                ts[i] = newIterator(def.c, start, end)
-                prefixLens[i] = len(def.prefix)
+                ts[i] = newIterator(def)
+                prefixLens[i] = def.prefixLen
                 defer ts[i].stop()

[dnj (Google), 2015/08/23 06:50:07]

This "i" is bound do the last value held by the iterator. You need to rebind in
the loop interior:
i := i

[iannucci, 2015/08/23 18:19:43]

good catch! Done.

         }
 
         suffix := []byte(nil)
         skip := -1
 
         for {
                 stop := false
                 restart := false
 
                 for idx, it := range ts {
@@ skipping 47 matching lines @@
         cb   func(*gkvlite.Item)
 }
 
 type iterator struct {
         stopper sync.Once
 
         stopped bool
         ch      chan<- *cmd
 }
 
-func newIterator(coll *memCollection, start, end []byte) *iterator {
+func newIterator(def *iterDefinition) *iterator {
         cmdChan := make(chan *cmd)
         ret := &iterator{
                 ch: cmdChan,
         }
 
+        // convert the suffixes from the iterDefinition into full rows for the
+        // underlying storage.
+        start := make([]byte, len(def.prefix)+len(def.start))
+        copy(start, def.prefix)

[dnj (Google), 2015/08/23 06:50:07]

Might as well use "bjoin" here.

[iannucci, 2015/08/23 18:19:43]

Oh ya!

+        copy(start[len(def.prefix):], def.start)
+
+        end := []byte(nil)
+        if def.end != nil {
+                end = make([]byte, len(def.prefix)+len(def.end))

[dnj (Google), 2015/08/23 06:50:07]

Might as well use "bjoin" here.

[iannucci, 2015/08/23 18:19:43]

Done.

+                copy(end, def.prefix)
+                copy(end[len(def.prefix):], def.end)
+        }
+
         go func() {
                 c := (*cmd)(nil)
                 ensureCmd := func() bool {
                         if c == nil {
                                 c = <-cmdChan
                                 if c == nil { // stop()
                                         return false
                                 }
                         }
                         return true
                 }
                 if ensureCmd() {
                         if bytes.Compare(c.targ, start) < 0 {
                                 c.targ = start
                         }
                 }
 
                 for {
+                        defer ret.stop()

[dnj (Google), 2015/08/23 06:50:07]

You should do this one before the loop.

[iannucci, 2015/08/23 18:19:43]

Done.

                         if !ensureCmd() {
                                 return
                         }
-                        needCallback := true
-                        coll.VisitItemsAscend(c.targ, true, func(i *gkvlite.Item) bool {
+                        terminalCallback := true
+                        def.c.VisitItemsAscend(c.targ, true, func(i *gkvlite.Item) bool {
                                 if !ensureCmd() {
                                         return false
                                 }
-                                if c.targ != nil && bytes.Compare(i.Key, c.targ) < 0 {
+                                if bytes.Compare(i.Key, c.targ) < 0 {
                                         // we need to start a new ascention function

[dnj (Google), 2015/08/23 06:50:07]

ascension

Is this actually a condition that can happen? It seems like gkvlite should
reliably avoid this possibility. If not, I'd kill this logic altogether, as it
would avoid a full byte compare per round and also remove the "terminalCallback"
logic.

[iannucci, 2015/08/23 18:19:43]

? This is when we have an iterator over

1
2
3
4
5

which we start at 1, but then subsequently ask for 4. It has nothing to do with
gkvlite?

-                                        needCallback = false
+                                        terminalCallback = false
+                                        return false
+                                }
+                                if !bytes.HasPrefix(i.Key, def.prefix) {
+                                        // we're no longer in prefix, terminate
                                         return false
                                 }
                                 if end != nil && bytes.Compare(i.Key, end) >= 0 {
-                                        // we hit our cap, return nil in callback outside loop.
+                                        // we hit our cap, terminate.
                                         return false
                                 }
                                 c.cb(i)
                                 c = nil
                                 return true
                         })
-                        if c != nil && needCallback {
+                        if terminalCallback && ensureCmd() {
                                 c.cb(nil)
                                 c = nil
                         }
                 }
         }()
 
         return ret
 }
 
 func (t *iterator) stop() {
@@ skipping 13 matching lines @@
         t.ch <- &cmd{targ, func(i *gkvlite.Item) {
                 defer close(waiter)
 
                 cb(i)
                 if i == nil {
                         t.stop()
                 }
         }}
         <-waiter
 }