From 0021854000693ebcfca2353dd5431f0435d574e8 Mon Sep 17 00:00:00 2001
From: "Sascha L. Teichmann" <sascha.teichmann@intevation.de>
Date: Mon, 1 Sep 2014 12:42:57 +0200
Subject: [PATCH] Added some spatial query for plain sql backend.

---
 common/coords.go | 28 +++++++++++++++
 sqlite.go        | 93 +++++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 120 insertions(+), 1 deletion(-)

diff --git a/common/coords.go b/common/coords.go
index 0bcf5aa..1efcf7b 100644
--- a/common/coords.go
+++ b/common/coords.go
@@ -37,6 +37,34 @@ func (c Coord) Equals(o Coord) bool {
 	return c.X == o.X && c.Y == o.Y && c.Z == o.Z
 }
 
+func minComponent(a, b int16) int16 {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func maxComponent(a, b int16) int16 {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+func MinCoord(a, b Coord) Coord {
+	return Coord{
+		X: minComponent(a.X, b.X),
+		Y: minComponent(a.Y, b.Y),
+		Z: minComponent(a.Z, b.Z)}
+}
+
+func MaxCoord(a, b Coord) Coord {
+	return Coord{
+		X: maxComponent(a.X, b.X),
+		Y: maxComponent(a.Y, b.Y),
+		Z: maxComponent(a.Z, b.Z)}
+}
+
 // Constructs a database key out of byte slice.
 func DecodeStringFromBytes(key []byte) (pos int64, err error) {
 	return strconv.ParseInt(string(key), 10, 64)
diff --git a/sqlite.go b/sqlite.go
index 1dd7f97..eb67ab8 100644
--- a/sqlite.go
+++ b/sqlite.go
@@ -23,6 +23,7 @@ const (
 	insertSql = "INSERT INTO blocks (pos, data) VALUES (?, ?)"
 	countSql  = "SELECT count(*) FROM blocks"
 	keysSql   = "SELECT pos FROM blocks"
+	rangeSql  = "SELECT pos, data FROM blocks WHERE pos BETWEEN ? AND ?"
 )
 
 type SqliteBackend struct {
@@ -36,6 +37,7 @@ type SqliteBackend struct {
 	updateStmt  *sql.Stmt
 	countStmt   *sql.Stmt
 	keysStmt    *sql.Stmt
+	rangeStmt   *sql.Stmt
 }
 
 type SqliteSession struct {
@@ -94,6 +96,11 @@ func NewSqliteBackend(path string, interleaved bool) (sqlb *SqliteBackend, err e
 		return
 	}
 
+	if res.rangeStmt, err = res.db.Prepare(rangeSql); err != nil {
+		res.closeAll()
+		return
+	}
+
 	if interleaved {
 		res.encoder = common.EncodeStringToBytesFromInterleaved
 		res.decoder = common.DecodeStringFromBytesToInterleaved
@@ -131,6 +138,7 @@ func (sqlb *SqliteBackend) closeAll() error {
 	closeStmt(&sqlb.existsStmt)
 	closeStmt(&sqlb.countStmt)
 	closeStmt(&sqlb.keysStmt)
+	closeStmt(&sqlb.rangeStmt)
 	return closeDB(&sqlb.db)
 }
 
@@ -276,6 +284,89 @@ func (ss *SqliteSession) AllKeys(hash []byte, done chan struct{}) (keys chan []b
 }
 
 func (ss *SqliteSession) SpatialQuery(hash, first, second []byte, done chan struct{}) (blocks chan Block, err error) {
-	err = ErrNotImplemented
+
+	// No implementation for the interleaved case, yet.
+	if ss.backend.interleaved {
+		err = ErrNotImplemented
+		return
+	}
+
+	return ss.PlainSpatialQuery(first, second, done)
+}
+
+func order(a, b int64) (int64, int64) {
+	if a < b {
+		return a, b
+	}
+	return b, a
+}
+
+func (ss *SqliteSession) PlainSpatialQuery(first, second []byte, done chan struct{}) (blocks chan Block, err error) {
+
+	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)
+
+	globalLock.RLock()
+
+	blocks = make(chan Block)
+
+	go func() {
+		defer globalLock.RUnlock()
+		defer close(blocks)
+		rangeStmt := ss.txStmt(ss.backend.rangeStmt)
+
+		a, b := common.Coord{X: c1.X}, common.Coord{X: c2.X}
+
+		for a.Z = c1.Z; a.Z <= c2.Z; a.Z++ {
+			b.Z = a.Z
+			for a.Y = c1.Y; a.Y <= c2.Y; a.Y++ {
+				b.Y = a.Y
+				var (
+					err  error
+					rows *sql.Rows
+				)
+				// Ordering should not be necessary.
+				from, to := order(common.CoordToPlain(a), common.CoordToPlain(b))
+				if rows, err = rangeStmt.Query(from, to); err != nil {
+					return
+				}
+				for rows.Next() {
+					var key int64
+					var data []byte
+					if err = rows.Scan(&key, &data); err != nil {
+						log.Printf("Error in range query: %s", err)
+						break
+					}
+					var encodedKey []byte
+					if encodedKey, err = common.EncodeStringToBytes(key); err != nil {
+						log.Printf("Key encoding failed: %s", err)
+						break
+					}
+					select {
+					case blocks <- Block{Key: encodedKey, Data: data}:
+					case <-done:
+						rows.Close()
+						return
+					}
+				}
+				if err = rows.Err(); err != nil {
+					log.Printf("Error in range query: %s", err)
+				}
+				rows.Close()
+			}
+		}
+	}()
+
 	return
 }