optimize reverse-sorted query

Author: Stebalien

datastore.go

@@ -6,7 +6,6 @@ import (
 
 	ds "github.com/ipfs/go-datastore"
 	dsq "github.com/ipfs/go-datastore/query"
-	"github.com/jbenet/goprocess"
 	"github.com/syndtr/goleveldb/leveldb"
 	"github.com/syndtr/goleveldb/leveldb/errors"
 	"github.com/syndtr/goleveldb/leveldb/iterator"
@@ -114,25 +113,29 @@ func (a *accessor) Delete(key ds.Key) (err error) {
 }
 
 func (a *accessor) Query(q dsq.Query) (dsq.Results, error) {
-	return a.queryNew(q)
-}
-
-func (a *accessor) queryNew(q dsq.Query) (dsq.Results, error) {
-	if len(q.Filters) > 0 ||
-		len(q.Orders) > 0 ||
-		q.Limit > 0 ||
-		q.Offset > 0 {
-		return a.queryOrig(q)
-	}
 	var rnge *util.Range
 	if q.Prefix != "" {
 		rnge = util.BytesPrefix([]byte(q.Prefix))
+		q.Prefix = ""
 	}
 	i := a.ldb.NewIterator(rnge, nil)
-	return dsq.ResultsFromIterator(q, dsq.Iterator{
+	next := i.Next
+	if len(q.Orders) > 0 {
+		switch q.Orders[0].(type) {
+		case dsq.OrderByKey, *dsq.OrderByKey:
+			q.Orders = nil
+		case dsq.OrderByKeyDescending, *dsq.OrderByKeyDescending:
+			next = func() bool {
+				next = i.Prev
+				return i.Last()
+			}
+			q.Orders = nil
+		default:
+		}
+	}
+	r := dsq.ResultsFromIterator(q, dsq.Iterator{
 		Next: func() (dsq.Result, bool) {
-			ok := i.Next()
-			if !ok {
+			if !next() {
 				return dsq.Result{}, false
 			}
 			k := string(i.Key())
@@ -149,86 +152,8 @@ func (a *accessor) queryNew(q dsq.Query) (dsq.Results, error) {
 			i.Release()
 			return nil
 		},
-	}), nil
-}
-
-func (a *accessor) queryOrig(q dsq.Query) (dsq.Results, error) {
-	// we can use multiple iterators concurrently. see:
-	// https://godoc.org/github.com/syndtr/goleveldb/leveldb#DB.NewIterator
-	// advance the iterator only if the reader reads
-	//
-	// run query in own sub-process tied to Results.Process(), so that
-	// it waits for us to finish AND so that clients can signal to us
-	// that resources should be reclaimed.
-	qrb := dsq.NewResultBuilder(q)
-	qrb.Process.Go(func(worker goprocess.Process) {
-		a.runQuery(worker, qrb)
 	})
-
-	// go wait on the worker (without signaling close)
-	go qrb.Process.CloseAfterChildren()
-
-	// Now, apply remaining things (filters, order)
-	qr := qrb.Results()
-	for _, f := range q.Filters {
-		qr = dsq.NaiveFilter(qr, f)
-	}
-	if len(q.Orders) > 0 {
-		switch q.Orders[0].(type) {
-		case dsq.OrderByKey, *dsq.OrderByKey:
-			// Default ordering
-		default:
-			qr = dsq.NaiveOrder(qr, q.Orders...)
-		}
-	}
-	return qr, nil
-}
-
-func (a *accessor) runQuery(worker goprocess.Process, qrb *dsq.ResultBuilder) {
-	var rnge *util.Range
-	if qrb.Query.Prefix != "" {
-		rnge = util.BytesPrefix([]byte(qrb.Query.Prefix))
-	}
-	i := a.ldb.NewIterator(rnge, nil)
-	defer i.Release()
-
-	// advance iterator for offset
-	if qrb.Query.Offset > 0 {
-		for j := 0; j < qrb.Query.Offset; j++ {
-			i.Next()
-		}
-	}
-
-	// iterate, and handle limit, too
-	for sent := 0; i.Next(); sent++ {
-		// end early if we hit the limit
-		if qrb.Query.Limit > 0 && sent >= qrb.Query.Limit {
-			break
-		}
-
-		k := string(i.Key())
-		e := dsq.Entry{Key: k}
-
-		if !qrb.Query.KeysOnly {
-			buf := make([]byte, len(i.Value()))
-			copy(buf, i.Value())
-			e.Value = buf
-		}
-
-		select {
-		case qrb.Output <- dsq.Result{Entry: e}: // we sent it out
-		case <-worker.Closing(): // client told us to end early.
-			break
-		}
-	}
-
-	if err := i.Error(); err != nil {
-		select {
-		case qrb.Output <- dsq.Result{Error: err}: // client read our error
-		case <-worker.Closing(): // client told us to end.
-			return
-		}
-	}
+	return dsq.NaiveQueryApply(q, r), nil
 }
 
 // DiskUsage returns the current disk size used by this levelDB.

go.mod

@@ -1,7 +1,6 @@
 module github.com/ipfs/go-ds-leveldb
 
 require (
-	github.com/ipfs/go-datastore v0.0.1
-	github.com/jbenet/goprocess v0.0.0-20160826012719-b497e2f366b8
+	github.com/ipfs/go-datastore v0.0.3
 	github.com/syndtr/goleveldb v1.0.0
 )

go.sum

@@ -10,8 +10,8 @@ github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
-github.com/ipfs/go-datastore v0.0.1 h1:AW/KZCScnBWlSb5JbnEnLKFWXL224LBEh/9KXXOrUms=
-github.com/ipfs/go-datastore v0.0.1/go.mod h1:d4KVXhMt913cLBEI/PXAy6ko+W7e9AhyAKBGh803qeE=
+github.com/ipfs/go-datastore v0.0.3 h1:/eP3nMDmLzMJNoWSSYvEkmMTTrm9FFCN+JraP9NdlwU=
+github.com/ipfs/go-datastore v0.0.3/go.mod h1:d4KVXhMt913cLBEI/PXAy6ko+W7e9AhyAKBGh803qeE=
 github.com/ipfs/go-ipfs-delay v0.0.0-20181109222059-70721b86a9a8/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw=
 github.com/jbenet/goprocess v0.0.0-20160826012719-b497e2f366b8 h1:bspPhN+oKYFk5fcGNuQzp6IGzYQSenLEgH3s6jkXrWw=
 github.com/jbenet/goprocess v0.0.0-20160826012719-b497e2f366b8/go.mod h1:Ly/wlsjFq/qrU3Rar62tu1gASgGw6chQbSh/XgIIXCY=