Adding caching for things stored as lists

tabular/gdc-graph/config.yaml

@@ -8,6 +8,7 @@ tables:
         elementList: "$.data.hits"
         params:
           size: "100"
+        cache: true
         pager:
           code: >
             def pager(url, result):

tabular/kvcache/index_ops.go

@@ -58,6 +58,27 @@ func (t *KVCache) NewLineIndex( path string ) (tabular.LineIndex, error) {
   return &LineIndex{t,num}, nil
 }
 
+func (t *KVCache) GetRowStorage(path string) (tabular.RowStorage, error) {
+  pk := PathKey(path)
+  v, err := t.KV.Get(pk)
+  if err != nil {
+    return nil, err
+  }
+  o, _ := binary.Uvarint(v)
+  return &KVRowStorage{t, o}, nil
+}
+
+func (t *KVCache) NewRowStorage( path string ) (tabular.RowStorage, error) {
+  ok := PathNumKey()
+  num := uint64(0)
+  if v, err := t.KV.Get(ok); err == nil {
+    num, _ = binary.Uvarint(v)
+  }
+  b := make([]byte, binary.MaxVarintLen64)
+  binary.PutUvarint(b, num+1)
+  t.KV.Set(ok, b) //Make part of same transaction as Get above?
+  return &KVRowStorage{t,num}, nil
+}
 
 
 func (t *LineIndex) AddIndexedField(colName string) {

tabular/kvcache/keys.go

@@ -12,6 +12,8 @@ var idPrefix   = []byte("g")
 var countPrefix = []byte("n")
 var idxPrefix   = []byte("x")
 
+var rowPrefix   = []byte("r")
+
 func PathNumKey() []byte {
 	return pathCount
 }
@@ -72,3 +74,31 @@ func LineCountKey(pathID uint64) []byte {
 	binary.PutUvarint(out[1:binary.MaxVarintLen64+1], pathID)
   return out
 }
+
+func RowPrefix(pathID uint64) []byte {
+  out := make([]byte, 1 + binary.MaxVarintLen64)
+  out[0] = rowPrefix[0]
+	binary.PutUvarint(out[1:binary.MaxVarintLen64+1], pathID)
+  return out
+}
+
+func RowKey(pathID uint64, id string) []byte {
+  p := []byte(id)
+  out := make([]byte, 1 + binary.MaxVarintLen64 + len(p))
+  out[0] = rowPrefix[0]
+	binary.PutUvarint(out[1:binary.MaxVarintLen64+1], pathID)
+  for i := 0; i < len(p); i++ {
+    out[i+1+binary.MaxVarintLen64] = p[i]
+  }
+  return out
+}
+
+func RowKeyParse(key []byte) (uint64, []byte) {
+	pathID, _ := binary.Uvarint(key[1:binary.MaxVarintLen64+1])
+	sLen := len(key) - (binary.MaxVarintLen64+1)
+	id := make([]byte, sLen)
+	for i := 0; i < sLen; i++ {
+		id[i] = key[i+1+binary.MaxVarintLen64]
+	}
+	return pathID, id
+}

tabular/kvcache/row_storage.go

@@ -0,0 +1,57 @@
+package kvcache
+
+import (
+  "log"
+  "context"
+  "bytes"
+  "encoding/json"
+  "github.com/bmeg/grip/tabular"
+  "github.com/bmeg/grip/kvi"
+)
+
+type KVRowStorage struct {
+  cache   *KVCache
+  pathID  uint64
+}
+
+
+func (r *KVRowStorage) Write(row *tabular.TableRow) error {
+  key := RowKey(r.pathID, row.Key)
+  buf := bytes.Buffer{}
+  enc := json.NewEncoder(&buf)
+  if err := enc.Encode(row.Values); err != nil {
+    log.Printf("Encode error: %s", err)
+  }
+  return r.cache.KV.Set(key, buf.Bytes())
+}
+
+
+func (r *KVRowStorage) GetRowsByField(ctx context.Context, field string, value string) chan *tabular.TableRow {
+  out := make(chan *tabular.TableRow, 10)
+  go func() {
+    defer close(out)
+    prefix := RowPrefix(r.pathID)
+    r.cache.KV.View(func(it kvi.KVIterator) error {
+      for it.Seek(prefix); it.Valid() && bytes.HasPrefix(it.Key(), prefix); it.Next() {
+        select {
+          case <-ctx.Done():
+            return nil
+          default:
+        }
+        _, id := RowKeyParse(it.Key())
+        data := map[string]interface{}{}
+        v, _ := it.Value()
+        buf := bytes.NewBuffer(v)
+        dec := json.NewDecoder(buf)
+        if err := dec.Decode(&data); err != nil {
+          log.Printf("Decode Error: %s", err)
+        }
+        if tabular.FieldFilter(field, value, data) {
+          out <- &tabular.TableRow{Key:string(id), Values:data}
+        }
+      }
+      return nil
+    })
+  }()
+  return out
+}

tabular/tabular.go

@@ -3,6 +3,8 @@ package tabular
 import (
   "fmt"
   "context"
+  "strings"
+  "github.com/oliveagle/jsonpath"
 )
 
 
@@ -42,10 +44,18 @@ type LineIndexWriter interface {
   IndexRow( line uint64, row map[string]interface{}) error
 }
 
+type RowStorage interface {
+  Write(row *TableRow) error
+  GetRowsByField(ctx context.Context, field string, value string) chan *TableRow
+}
+
 //Cache
 type Cache interface {
   NewLineIndex(path string) (LineIndex, error)
   GetLineIndex(path string) (LineIndex, error)
+
+  NewRowStorage(path string) (RowStorage, error)
+  GetRowStorage(path string) (RowStorage, error)
 }
 
 type Options struct {
@@ -84,3 +94,27 @@ func (t *TableManager) NewDriver(name string, url string, opts Options) (Driver,
 func NewCache(path string) (Cache, error) {
   return cacheMap["kv"](path)
 }
+
+
+func pathFix(p string) string {
+  if !strings.HasPrefix(p, "$.") {
+    return "$." + p
+  }
+  return p
+}
+
+func FieldFilter(field string, value string, data map[string]interface{}) bool {
+  if field == "" {
+    return true
+  }
+  v, err := jsonpath.JsonPathLookup(data, pathFix(field) )
+  if err != nil {
+    return false
+  }
+  if valStr, ok := v.(string); ok {
+    if valStr == value {
+      return true
+    }
+  }
+  return false
+}

tabular/web/driver.go

@@ -18,6 +18,8 @@ import (
 type Driver struct {
   conf Config
   opts tabular.Options
+  cache      tabular.Cache
+  rowStorage tabular.RowStorage
 }
 
 type QueryConfig struct {
@@ -26,6 +28,7 @@ type QueryConfig struct {
   Element     string  `json:"element"`
   Params      map[string]string `json:"params"`
   Headers     []string          `json:"headers"`
+  Cache       bool              `json:"cache"`
 }
 
 type Config struct {
@@ -34,7 +37,7 @@ type Config struct {
 }
 
 func TSVDriverBuilder(url string, manager tabular.Cache, opts tabular.Options) (tabular.Driver, error) {
-  o := Driver{opts:opts}
+  o := Driver{opts:opts, cache:manager}
   conf := Config{}
   err := mapstructure.Decode(opts.Config, &conf)
   if err != nil {
@@ -60,7 +63,21 @@ func pathFix(p string) string {
   return p
 }
 
-func (d *Driver) GetRows(ctx context.Context) chan *tabular.TableRow {
+func (d *Driver) buildCache() {
+  if d.conf.List != nil && d.conf.List.Cache {
+    url := d.conf.List.URL //BUG: need to make sure URL is unique in config file
+    var err error
+    if d.rowStorage, err = d.cache.GetRowStorage(url); err != nil {
+      d.rowStorage, _ = d.cache.NewRowStorage(url)
+    }
+    log.Printf("Caching %s", d.conf.List.URL)
+    for row := range d.fetchRows(context.TODO()) {
+      d.rowStorage.Write(row)
+    }
+  }
+}
+
+func (d *Driver) fetchRows(ctx context.Context) chan *tabular.TableRow {
   out := make(chan *tabular.TableRow, 10)
   go func() {
     defer close(out)
@@ -110,10 +127,17 @@ func (d *Driver) GetRows(ctx context.Context) chan *tabular.TableRow {
       }
     }
   }()
-
   return out
 }
 
+func (d *Driver) GetRows(ctx context.Context) chan *tabular.TableRow {
+  d.buildCache()
+  if d.rowStorage != nil {
+    return d.rowStorage.GetRowsByField(ctx, "", "")
+  }
+  return d.fetchRows(ctx)
+}
+
 func (d *Driver) GetRowByID(id string) (*tabular.TableRow, error) {
   log.Printf("Getting row: %s", id)
   if tableGet, ok := d.conf.Get[d.opts.PrimaryKey]; ok {