From f4baf63247175fd6ba1d7a48d277a53bb3e2686e Mon Sep 17 00:00:00 2001
From: "Sascha L. Teichmann" <sascha.teichmann@intevation.de>
Date: Mon, 1 Sep 2014 18:04:48 +0200
Subject: [PATCH] Added interleved implementation of spatial query in LevelDB
 backend.

---
 leveldb.go | 76 +++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 75 insertions(+), 1 deletion(-)

diff --git a/leveldb.go b/leveldb.go
index 058a88c..3e1cf03 100644
--- a/leveldb.go
+++ b/leveldb.go
@@ -303,6 +303,80 @@ func (ldbs *LevelDBSession) plainSpatialQuery(first, second []byte, done chan st
 }
 
 func (ldbs *LevelDBSession) interleavedSpatialQuery(first, second []byte, done chan struct{}) (blocks chan Block, err error) {
-	err = ErrNotImplemented
+	var (
+		firstKey  int64
+		secondKey int64
+	)
+	if firstKey, err = common.DecodeStringFromBytes(first); err != nil {
+		return
+	}
+	if secondKey, err = common.DecodeStringFromBytes(second); err != nil {
+		return
+	}
+	c1 := common.PlainToCoord(firstKey)
+	c2 := common.PlainToCoord(secondKey)
+	c1, c2 = common.MinCoord(c1, c2), common.MaxCoord(c1, c2)
+
+	blocks = make(chan Block)
+
+	go func() {
+		defer close(blocks)
+		ldbs.backend.mutex.RLock()
+		defer ldbs.backend.mutex.RUnlock()
+
+		ro := leveldb.NewReadOptions()
+		defer ro.Close()
+		ro.SetFillCache(false)
+
+		it := ldbs.backend.db.NewIterator(ro)
+		defer it.Close()
+
+		zmin, zmax := common.CoordToInterleaved(c1), common.CoordToInterleaved(c2)
+		// Should not be necessary.
+		zmin, zmax = order(zmin, zmax)
+		var (
+			cub        = common.Cuboid{P1: c1, P2: c2}
+			zcode      = zmin
+			err        error
+			encodedKey []byte
+		)
+	SEEK:
+		if encodedKey, err = common.EncodeToBigEndian(zcode); err != nil {
+			log.Printf("error encoding key: %s", err)
+			return
+		}
+		it.Seek(encodedKey)
+		for it.Valid() {
+			if zcode, err = common.DecodeFromBigEndian(it.Key()); err != nil {
+				log.Printf("error decoding key: %s", err)
+				return
+			}
+
+			if zcode > zmax {
+				break
+			}
+
+			if c := common.InterleavedToCoord(zcode); cub.Contains(c) {
+				if encodedKey, err = common.EncodeStringToBytes(common.CoordToPlain(c)); err != nil {
+					log.Printf("error encoding key: %s", err)
+					return
+				}
+				select {
+				case blocks <- Block{Key: encodedKey, Data: it.Value()}:
+				case <-done:
+					return
+				}
+			} else {
+				zmin = common.NaiveBigMin(zmin, zmax, zcode)
+				goto SEEK
+			}
+
+			it.Next()
+		}
+		if err = it.GetError(); err != nil {
+			log.Printf("error while iterating: %s", err)
+			return
+		}
+	}()
 	return
 }