scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add time-based frames and Range() support.

Browse source
This commit is contained in:
Ben Johnson 2016-12-08 12:29:49 -07:00
parent ee7240107c
commit c6e2b1294b
23 changed files with 4561 additions and 833 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
attr.go
View file

@ -17,6 +17,13 @@ import (
// AttrBlockSize is the size of attribute blocks for anti-entropy.
const AttrBlockSize = 100
// Attribute data type enum.
const (
AttrTypeString = 1
AttrTypeUint = 2
AttrTypeBool = 3
)
// AttrStore represents a storage layer for attributes.
type AttrStore struct {
mu sync.Mutex
@ -303,32 +310,39 @@ func decodeAttrs(pb []*internal.Attr) map[string]interface{} {
// encodeAttr converts a key/value pair into an Attr internal representation.
func encodeAttr(key string, value interface{}) *internal.Attr {
pb := &internal.Attr{Key: proto.String(key)}
pb := &internal.Attr{Key: key}
switch value := value.(type) {
case string:
pb.StringValue = proto.String(value)
pb.Type = AttrTypeString
pb.StringValue = value
case float64:
pb.UintValue = proto.Uint64(uint64(value))
pb.Type = AttrTypeUint
pb.UintValue = uint64(value)
case uint64:
pb.UintValue = proto.Uint64(value)
pb.Type = AttrTypeUint
pb.UintValue = value
case int64:
pb.UintValue = proto.Uint64(uint64(value))
pb.Type = AttrTypeUint
pb.UintValue = uint64(value)
case bool:
pb.BoolValue = proto.Bool(value)
pb.Type = AttrTypeBool
pb.BoolValue = value
}
return pb
}
// decodeAttr converts from an Attr internal representation to a key/value pair.
func decodeAttr(attr *internal.Attr) (key string, value interface{}) {
if attr.StringValue != nil {
return attr.GetKey(), attr.GetStringValue()
} else if attr.UintValue != nil {
return attr.GetKey(), attr.GetUintValue()
} else if attr.BoolValue != nil {
return attr.GetKey(), attr.GetBoolValue()
switch attr.Type {
case AttrTypeString:
return attr.Key, attr.StringValue
case AttrTypeUint:
return attr.Key, attr.UintValue
case AttrTypeBool:
return attr.Key, attr.BoolValue
default:
return attr.Key, nil
}
return attr.GetKey(), nil
}
// cloneAttrs returns a shallow clone of m.

4
bitmap.go
View file

@ -227,8 +227,8 @@ func decodeBitmap(pb *internal.Bitmap) *Bitmap {
}
b := NewBitmap()
b.Attrs = decodeAttrs(pb.GetAttrs())
for _, v := range pb.GetBits() {
b.Attrs = decodeAttrs(pb.Attrs)
for _, v := range pb.Bits {
b.SetBit(v)
}
return b

9
cache.go
View file

@ -7,7 +7,6 @@ import (
"sort"
"time"
"github.com/gogo/protobuf/proto"
"github.com/golang/groupcache/lru"
"github.com/pilosa/pilosa/internal"
)
@ -216,15 +215,15 @@ type Pair struct {
func encodePair(p Pair) *internal.Pair {
return &internal.Pair{
Key: proto.Uint64(p.Key),
Count: proto.Uint64(p.Count),
Key: p.Key,
Count: p.Count,
}
}
func decodePair(pb *internal.Pair) Pair {
return Pair{
Key: pb.GetKey(),
Count: pb.GetCount(),
Key: pb.Key,
Count: pb.Count,
}
}

24
client.go
View file

@ -151,9 +151,9 @@ func (c *Client) ExecuteQuery(ctx context.Context, db, query string, allowRedire
// Encode query request.
buf, err := proto.Marshal(&internal.QueryRequest{
DB: proto.String(db),
Query: proto.String(query),
Remote: proto.Bool(!allowRedirect),
DB: db,
Query: query,
Remote: !allowRedirect,
})
if err != nil {
return nil, fmt.Errorf("marshal: %s", err)
@ -187,7 +187,7 @@ func (c *Client) ExecuteQuery(ctx context.Context, db, query string, allowRedire
var qresp internal.QueryResponse
if err := proto.Unmarshal(body, &qresp); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
} else if s := qresp.GetErr(); s != "" {
} else if s := qresp.Err; s != "" {
return nil, errors.New(s)
}
@ -230,9 +230,9 @@ func MarshalImportPayload(db, frame string, slice uint64, bits []Bit) ([]byte, e
// Marshal bits to protobufs.
buf, err := proto.Marshal(&internal.ImportRequest{
DB: proto.String(db),
Frame: proto.String(frame),
Slice: proto.Uint64(slice),
DB: db,
Frame: frame,
Slice: slice,
BitmapIDs: bitmapIDs,
ProfileIDs: profileIDs,
})
@ -272,7 +272,7 @@ func (c *Client) importNode(ctx context.Context, node *Node, buf []byte) error {
var isresp internal.ImportResponse
if err := proto.Unmarshal(body, &isresp); err != nil {
return fmt.Errorf("unmarshal import response: %s", err)
} else if s := isresp.GetErr(); s != "" {
} else if s := isresp.Err; s != "" {
return errors.New(s)
}
@ -644,10 +644,10 @@ func (c *Client) FragmentBlocks(ctx context.Context, db, frame string, slice uin
// BlockData returns bitmap/profile id pairs for a block.
func (c *Client) BlockData(ctx context.Context, db, frame string, slice uint64, block int) ([]uint64, []uint64, error) {
buf, err := proto.Marshal(&internal.BlockDataRequest{
DB: proto.String(db),
Frame: proto.String(frame),
Slice: proto.Uint64(slice),
Block: proto.Uint64(uint64(block)),
DB: db,
Frame: frame,
Slice: slice,
Block: uint64(block),
})
if err != nil {
return nil, nil, err

6
client_test.go
View file

@ -96,11 +96,11 @@ func TestClient_FragmentBlocks(t *testing.T) {
defer idx.Close()
// Set two bits on blocks 0 & 3.
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(pilosa.HashBlockSize*3, 100, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(pilosa.HashBlockSize*3, 100)
// Set a bit on a different slice.
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, 1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, 1)
s := NewServer()
defer s.Close()

149
db.go
View file

@ -3,10 +3,15 @@ package pilosa
import (
"errors"
"fmt"
"io/ioutil"
"os"
"path/filepath"
"sort"
"sync"
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
)
// DB represents a container for frames.
@ -15,6 +20,10 @@ type DB struct {
path string
name string
// Default time quantum for all frames in database.
// This can be overridden by individual frames.
timeQuantum TimeQuantum
// Frames by name.
frames map[string]*Frame
@ -53,6 +62,11 @@ func (db *DB) Open() error {
return err
}
// Read meta file.
if err := db.loadMeta(); err != nil {
return err
}
if err := db.openFrames(); err != nil {
return err
}
@ -93,6 +107,45 @@ func (db *DB) openFrames() error {
return nil
}
// loadMeta reads meta data for the database, if any.
func (db *DB) loadMeta() error {
var pb internal.DB
// Read data from meta file.
buf, err := ioutil.ReadFile(filepath.Join(db.path, "meta"))
if os.IsNotExist(err) {
db.timeQuantum = ""
return nil
} else if err != nil {
return err
} else {
if err := proto.Unmarshal(buf, &pb); err != nil {
return err
}
}
// Copy metadata fields.
db.timeQuantum = TimeQuantum(pb.TimeQuantum)
return nil
}
// saveMeta writes meta data for the database.
func (db *DB) saveMeta() error {
// Marshal metadata.
buf, err := proto.Marshal(&internal.DB{TimeQuantum: string(db.timeQuantum)})
if err != nil {
return err
}
// Write to meta file.
if err := ioutil.WriteFile(filepath.Join(db.path, "meta"), buf, 0666); err != nil {
return err
}
return nil
}
// Close closes the database and its frames.
func (db *DB) Close() error {
db.mu.Lock()
@ -126,6 +179,34 @@ func (db *DB) SliceN() uint64 {
return max
}
// TimeQuantum returns the default time quantum for the database.
func (db *DB) TimeQuantum() TimeQuantum {
db.mu.Lock()
defer db.mu.Unlock()
return db.timeQuantum
}
// SetTimeQuantum sets the default time quantum for the database.
func (db *DB) SetTimeQuantum(q TimeQuantum) error {
db.mu.Lock()
defer db.mu.Unlock()
// Validate input.
if !q.Valid() {
return ErrInvalidTimeQuantum
}
// Update value on database.
db.timeQuantum = q
// Perist meta data to disk.
if err := db.saveMeta(); err != nil {
return err
}
return nil
}
// FramePath returns the path to a frame in the database.
func (db *DB) FramePath(name string) string { return filepath.Join(db.path, name) }
@ -187,6 +268,74 @@ func (db *DB) newFrame(path, name string) *Frame {
return f
}
// DeleteFrame removes a frame from the database.
func (db *DB) DeleteFrame(name string) error {
db.mu.Lock()
defer db.mu.Unlock()
// Ignore if frame doesn't exist.
f := db.frame(name)
if f == nil {
return nil
}
// Close frame.
if err := f.Close(); err != nil {
return err
}
// Delete frame directory.
if err := os.RemoveAll(db.FramePath(name)); err != nil {
return err
}
// Remove reference.
delete(db.frames, name)
db.stats.Count("frameN", -1)
return nil
}
// SetBit sets a bit for a given profile & bitmap.
// If a timestamp is specified then set all bits for the different quantum units.
func (db *DB) SetBit(name string, bitmapID, profileID uint64, t *time.Time) (changed bool, err error) {
// Read frame.
f, err := db.CreateFrameIfNotExists(name)
if err != nil {
return changed, err
}
// If this is a non-time bit then simply set the bit on the frame.
if t == nil {
return f.SetBit(bitmapID, profileID)
}
// Determine quantum of frame. Set to the default quantum if it is unset.
q := f.TimeQuantum()
if q == "" {
q = db.TimeQuantum()
if err := f.SetTimeQuantum(q); err != nil {
return changed, err
}
}
// If a timestamp is specified then set bits across all frames for the quantum.
for _, subname := range FramesByTime(name, *t, q) {
f, err := db.CreateFrameIfNotExists(subname)
if err != nil {
return changed, err
}
if c, err := f.SetBit(bitmapID, profileID); err != nil {
return changed, err
} else if c {
changed = true
}
}
return changed, nil
}
type dbSlice []*DB
func (p dbSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }

68
db_test.go
View file

@ -4,6 +4,7 @@ import (
"io/ioutil"
"os"
"testing"
"time"
"github.com/pilosa/pilosa"
)
@ -34,6 +35,49 @@ func TestDB_CreateFrameIfNotExists(t *testing.T) {
}
}
// Ensure database can delete a frame.
func TestDB_DeleteFrame(t *testing.T) {
db := MustOpenDB()
defer db.Close()
// Create frame.
if _, err := db.CreateFrameIfNotExists("f"); err != nil {
t.Fatal(err)
}
// Delete frame & verify it's gone.
if err := db.DeleteFrame("f"); err != nil {
t.Fatal(err)
} else if db.Frame("f") != nil {
t.Fatal("expected nil frame")
}
// Delete again to make sure it doesn't error.
if err := db.DeleteFrame("f"); err != nil {
t.Fatal(err)
}
}
// Ensure database can set the default time quantum.
func TestDB_SetTimeQuantum(t *testing.T) {
db := MustOpenDB()
defer db.Close()
// Set & retrieve time quantum.
if err := db.SetTimeQuantum(pilosa.TimeQuantum("YMDH")); err != nil {
t.Fatal(err)
} else if q := db.TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum: %s", q)
}
// Reload database and verify that it is persisted.
if err := db.Reopen(); err != nil {
t.Fatal(err)
} else if q := db.TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum (reopen): %s", q)
}
}
// DB represents a test wrapper for pilosa.DB.
type DB struct {
*pilosa.DB
@ -63,3 +107,27 @@ func (db *DB) Close() error {
defer os.RemoveAll(db.Path())
return db.DB.Close()
}
// Reopen closes the database and reopens it.
func (db *DB) Reopen() error {
if err := db.DB.Close(); err != nil {
return err
}
path, name := db.Path(), db.Name()
db.DB = pilosa.NewDB(path, name)
if err := db.Open(); err != nil {
return err
}
return nil
}
// MustSetBit sets a bit on the database. Panic on error.
func (db *DB) MustSetBit(name string, bitmapID, profileID uint64, t *time.Time) (changed bool) {
changed, err := db.SetBit(name, bitmapID, profileID, t)
if err != nil {
panic(err)
}
return changed
}

52
executor.go
View file

@ -304,11 +304,28 @@ func (e *Executor) executeRangeSlice(ctx context.Context, db string, c *pql.Rang
frame = DefaultFrame
}
f := e.Index.Fragment(db, frame, slice)
// Retrieve base frame.
f := e.Index.Frame(db, frame)
if f == nil {
return NewBitmap(), nil
return &Bitmap{}, nil
}
return f.Range(c.ID, c.StartTime, c.EndTime), nil
// If no quantum exists then return an empty bitmap.
q := f.TimeQuantum()
if q == "" {
return &Bitmap{}, nil
}
// Union bitmaps across all time-based subframes.
bm := &Bitmap{}
for _, subframe := range FramesByTimeRange(frame, c.StartTime, c.EndTime, q) {
f := e.Index.Fragment(db, subframe, slice)
if f == nil {
continue
}
bm = bm.Union(f.Bitmap(c.ID))
}
return bm, nil
}
// executeUnionSlice executes a union() call for a local slice.
@ -405,15 +422,14 @@ func (e *Executor) executeSetBit(ctx context.Context, db string, c *pql.SetBit,
for _, node := range e.Cluster.FragmentNodes(db, slice) {
// Update locally if host matches.
if node.Host == e.Host {
f, err := e.Index.CreateFragmentIfNotExists(db, c.Frame, slice)
db, err := e.Index.CreateDBIfNotExists(db)
if err != nil {
return false, fmt.Errorf("fragment: %s", err)
return false, fmt.Errorf("db: %s", err)
}
val, err := f.SetBit(c.ID, c.ProfileID, opt.Timestamp, opt.Quantum)
val, err := db.SetBit(c.Frame, c.ID, c.ProfileID, opt.Timestamp)
if err != nil {
return false, err
}
if val {
} else if val {
ret = true
}
continue
@ -579,14 +595,13 @@ func (e *Executor) executeSetProfileAttrs(ctx context.Context, db string, c *pql
func (e *Executor) exec(ctx context.Context, node *Node, db string, q *pql.Query, slices []uint64, opt *ExecOptions) (results []interface{}, err error) {
// Encode request object.
pbreq := &internal.QueryRequest{
DB: proto.String(db),
Query: proto.String(q.String()),
Slices: slices,
Quantum: proto.Uint32(uint32(opt.Quantum)),
Remote: proto.Bool(true),
DB: db,
Query: q.String(),
Slices: slices,
Remote: true,
}
if opt.Timestamp != nil {
pbreq.Timestamp = proto.Int64(opt.Timestamp.UnixNano())
pbreq.Timestamp = opt.Timestamp.UnixNano()
}
buf, err := proto.Marshal(pbreq)
if err != nil {
@ -632,7 +647,7 @@ func (e *Executor) exec(ctx context.Context, node *Node, db string, q *pql.Query
}
// Return an error, if specified on response.
if err := decodeError(pb.GetErr()); err != nil {
if err := decodeError(pb.Err); err != nil {
return nil, err
}
@ -648,11 +663,11 @@ func (e *Executor) exec(ctx context.Context, node *Node, db string, q *pql.Query
case *pql.TopN:
v, err = decodePairs(pb.Results[i].GetPairs()), nil
case *pql.Count:
v, err = pb.Results[i].GetN(), nil
v, err = pb.Results[i].N, nil
case *pql.SetBit:
v, err = pb.Results[i].GetChanged(), nil
v, err = pb.Results[i].Changed, nil
case *pql.ClearBit:
v, err = pb.Results[i].GetChanged(), nil
v, err = pb.Results[i].Changed, nil
case *pql.SetBitmapAttrs:
case *pql.SetProfileAttrs:
default:
@ -837,7 +852,6 @@ type mapResponse struct {
// ExecOptions represents an execution context for a single Execute() call.
type ExecOptions struct {
Timestamp *time.Time
Quantum TimeQuantum
Remote bool
}

72
executor_test.go
View file

@ -171,15 +171,15 @@ func TestExecutor_Execute_TopN(t *testing.T) {
defer idx.Close()
// Set bits for bitmaps 0, 10, & 20 across two slices.
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth+2, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 5).SetBit(0, (5*SliceWidth)+100, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(10, 0, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(20, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "other", 0).SetBit(0, 0, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth+2)
idx.MustCreateFragmentIfNotExists("d", "f", 5).SetBit(0, (5*SliceWidth)+100)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(10, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(20, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "other", 0).SetBit(0, 0)
// Execute query.
e := NewExecutor(idx.Index, NewCluster(1))
@ -197,12 +197,12 @@ func TestExecutor_Execute_TopN_fill(t *testing.T) {
defer idx.Close()
// Set bits for bitmaps 0, 10, & 20 across two slices.
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 2, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(1, SliceWidth+2, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(1, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 2)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(1, SliceWidth+2)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(1, SliceWidth)
// Execute query.
e := NewExecutor(idx.Index, NewCluster(1))
@ -221,19 +221,19 @@ func TestExecutor_Execute_TopN_Src(t *testing.T) {
defer idx.Close()
// Set bits for bitmaps 0, 10, & 20 across two slices.
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth+1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(20, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(20, SliceWidth+1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(20, SliceWidth+2, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(20, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(20, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(20, SliceWidth+2)
// Create an intersecting bitmap.
idx.MustCreateFragmentIfNotExists("d", "other", 1).SetBit(100, SliceWidth, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "other", 1).SetBit(100, SliceWidth+1, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "other", 1).SetBit(100, SliceWidth+2, nil, 0)
idx.MustCreateFragmentIfNotExists("d", "other", 1).SetBit(100, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "other", 1).SetBit(100, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "other", 1).SetBit(100, SliceWidth+2)
// Execute query.
e := NewExecutor(idx.Index, NewCluster(1))
@ -253,15 +253,23 @@ func TestExecutor_Execute_Range(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
f := idx.MustCreateFragmentIfNotExists("d", "f.t", 0)
if _, err := f.SetBit(1, 100, MustParseTime("2000-01-01 00:00"), pilosa.YMD); err != nil {
t.Fatal(err)
}
db := idx.MustCreateDBIfNotExists("d")
db.SetTimeQuantum(pilosa.TimeQuantum("YMDH"))
db.MustSetBit("f", 1, 2, MustParseTimePtr("1999-12-31 00:00"))
db.MustSetBit("f", 1, 3, MustParseTimePtr("2000-01-01 00:00"))
db.MustSetBit("f", 1, 4, MustParseTimePtr("2000-01-02 00:00"))
db.MustSetBit("f", 1, 5, MustParseTimePtr("2000-02-01 00:00"))
db.MustSetBit("f", 1, 6, MustParseTimePtr("2001-01-01 00:00"))
db.MustSetBit("f", 1, 7, MustParseTimePtr("2002-01-01 02:00"))
db.MustSetBit("f", 1, 2, MustParseTimePtr("1999-12-30 00:00")) // too early
db.MustSetBit("f", 1, 2, MustParseTimePtr("2002-02-01 00:00")) // too late
db.MustSetBit("f", 10, 2, MustParseTimePtr("2001-01-01 00:00")) // different bitmap
e := NewExecutor(idx.Index, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Range(id=1, frame=f.t, start="2000-01-01T00:00", end="2000-01-01T01:00")`), nil, nil); err != nil {
if res, err := e.Execute(context.Background(), "d", MustParse(`Range(id=1, frame=f, start="1999-12-31T00:00", end="2002-01-01T03:00")`), nil, nil); err != nil {
t.Fatal(err)
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{100}) {
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{2, 3, 4, 5, 6, 7}) {
t.Fatalf("unexpected bits: %+v", bits)
}
}

44
fragment.go
View file

@ -238,7 +238,7 @@ func (f *Fragment) openCache() error {
// Read in all bitmaps by ID.
// This will cause them to be added to the cache.
for _, bitmapID := range pb.GetBitmapIDs() {
for _, bitmapID := range pb.BitmapIDs {
n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth)
f.cache.Add(bitmapID, n)
}
@ -323,11 +323,6 @@ func (f *Fragment) bitmap(bitmapID uint64) *Bitmap {
}
bm.InvalidateCount()
// f.storage.ForEachRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth, func(i uint64) {
// profileID := (f.slice * SliceWidth) + (i % SliceWidth)
// bm.SetBit(profileID)
// })
// Update cache.
f.cache.Add(bitmapID, bm.Count())
@ -336,15 +331,9 @@ func (f *Fragment) bitmap(bitmapID uint64) *Bitmap {
// SetBit sets a bit for a given profile & bitmap within the fragment.
// This updates both the on-disk storage and the in-cache bitmap.
func (f *Fragment) SetBit(bitmapID, profileID uint64, t *time.Time, q TimeQuantum) (changed bool, err error) {
func (f *Fragment) SetBit(bitmapID, profileID uint64) (changed bool, err error) {
f.mu.Lock()
defer f.mu.Unlock()
// Set time bits if this is a time-frame and a timestamp is specified.
if strings.HasSuffix(f.frame, FrameSuffixTime) && t != nil {
return f.setTimeBit(bitmapID, profileID, *t, q)
}
return f.setBit(bitmapID, profileID)
}
@ -379,17 +368,6 @@ func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error)
return changed, nil
}
func (f *Fragment) setTimeBit(bitmapID, profileID uint64, t time.Time, q TimeQuantum) (changed bool, err error) {
for _, timeID := range TimeIDsFromQuantum(q, t, bitmapID) {
if v, err := f.setBit(timeID, profileID); err != nil {
return changed, fmt.Errorf("set time bit: t=%s, q=%s, err=%s", t, q, err)
} else if v {
changed = true
}
}
return changed, nil
}
// ClearBit clears a bit for a given profile & bitmap within the fragment.
// This updates both the on-disk storage and the in-cache bitmap.
func (f *Fragment) ClearBit(bitmapID, profileID uint64) (bool, error) {
@ -601,24 +579,6 @@ type TopOptions struct {
FilterValues []interface{}
}
func (f *Fragment) Range(bitmapID uint64, start, end time.Time) *Bitmap {
f.mu.Lock()
defer f.mu.Unlock()
// Retrieve a list of bitmap ids for a given time range.
bitmapIDs := TimeIDsFromRange(start, end, bitmapID)
if len(bitmapIDs) == 0 {
return NewBitmap()
}
// Union all bitmap ids from the time range.
bm := f.bitmap(bitmapIDs[0])
for _, id := range bitmapIDs[1:] {
bm = bm.Union(f.bitmap(id))
}
return bm
}
// Checksum returns a checksum for the entire fragment.
// If two fragments have the same checksum then they have the same data.
func (f *Fragment) Checksum() []byte {

46
fragment_test.go
View file

@ -27,11 +27,11 @@ func TestFragment_SetBit(t *testing.T) {
defer f.Close()
// Set bits on the fragment.
if _, err := f.SetBit(120, 1, nil, 0); err != nil {
if _, err := f.SetBit(120, 1); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(120, 6, nil, 0); err != nil {
} else if _, err := f.SetBit(120, 6); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(121, 0, nil, 0); err != nil {
} else if _, err := f.SetBit(121, 0); err != nil {
t.Fatal(err)
}
@ -58,9 +58,9 @@ func TestFragment_ClearBit(t *testing.T) {
defer f.Close()
// Set and then clear bits on the fragment.
if _, err := f.SetBit(1000, 1, nil, 0); err != nil {
if _, err := f.SetBit(1000, 1); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(1000, 2, nil, 0); err != nil {
} else if _, err := f.SetBit(1000, 2); err != nil {
t.Fatal(err)
} else if _, err := f.ClearBit(1000, 1); err != nil {
t.Fatal(err)
@ -85,9 +85,9 @@ func TestFragment_Snapshot(t *testing.T) {
defer f.Close()
// Set and then clear bits on the fragment.
if _, err := f.SetBit(1000, 1, nil, 0); err != nil {
if _, err := f.SetBit(1000, 1); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(1000, 2, nil, 0); err != nil {
} else if _, err := f.SetBit(1000, 2); err != nil {
t.Fatal(err)
} else if _, err := f.ClearBit(1000, 1); err != nil {
t.Fatal(err)
@ -114,11 +114,11 @@ func TestFragment_ForEachBit(t *testing.T) {
defer f.Close()
// Set bits on the fragment.
if _, err := f.SetBit(100, 20, nil, 0); err != nil {
if _, err := f.SetBit(100, 20); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(2, 38, nil, 0); err != nil {
} else if _, err := f.SetBit(2, 38); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(2, 37, nil, 0); err != nil {
} else if _, err := f.SetBit(2, 37); err != nil {
t.Fatal(err)
}
@ -284,9 +284,9 @@ func TestFragment_Checksum(t *testing.T) {
// Retrieve checksum and set bits.
orig := f.Checksum()
if _, err := f.SetBit(1, 200, nil, 0); err != nil {
if _, err := f.SetBit(1, 200); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.HashBlockSize*2, 200, nil, 0); err != nil {
} else if _, err := f.SetBit(pilosa.HashBlockSize*2, 200); err != nil {
t.Fatal(err)
}
@ -305,7 +305,7 @@ func TestFragment_Blocks(t *testing.T) {
var prev []pilosa.FragmentBlock
// Set first bit.
if _, err := f.SetBit(0, 0, nil, 0); err != nil {
if _, err := f.SetBit(0, 0); err != nil {
t.Fatal(err)
}
blocks := f.Blocks()
@ -315,7 +315,7 @@ func TestFragment_Blocks(t *testing.T) {
prev = blocks
// Set bit on different bitmap.
if _, err := f.SetBit(20, 0, nil, 0); err != nil {
if _, err := f.SetBit(20, 0); err != nil {
t.Fatal(err)
}
blocks = f.Blocks()
@ -325,7 +325,7 @@ func TestFragment_Blocks(t *testing.T) {
prev = blocks
// Set bit on different profile.
if _, err := f.SetBit(20, 100, nil, 0); err != nil {
if _, err := f.SetBit(20, 100); err != nil {
t.Fatal(err)
}
blocks = f.Blocks()
@ -340,7 +340,7 @@ func TestFragment_Blocks_Empty(t *testing.T) {
defer f.Close()
// Set bits on a different block.
if _, err := f.SetBit(100, 1, nil, 0); err != nil {
if _, err := f.SetBit(100, 1); err != nil {
t.Fatal(err)
}
@ -359,7 +359,7 @@ func TestFragment_LRUCache_Persistence(t *testing.T) {
// Set bits on the fragment.
for i := uint64(0); i < 1000; i++ {
if _, err := f.SetBit(i, 0, nil, 0); err != nil {
if _, err := f.SetBit(i, 0); err != nil {
t.Fatal(err)
}
}
@ -391,7 +391,7 @@ func TestFragment_RankCache_Persistence(t *testing.T) {
// Set bits on the fragment.
for i := uint64(0); i < 1000; i++ {
if _, err := f.SetBit(i, 0, nil, 0); err != nil {
if _, err := f.SetBit(i, 0); err != nil {
t.Fatal(err)
}
}
@ -422,9 +422,9 @@ func TestFragment_WriteTo_ReadFrom(t *testing.T) {
defer f0.Close()
// Set and then clear bits on the fragment.
if _, err := f0.SetBit(1000, 1, nil, 0); err != nil {
if _, err := f0.SetBit(1000, 1); err != nil {
t.Fatal(err)
} else if _, err := f0.SetBit(1000, 2, nil, 0); err != nil {
} else if _, err := f0.SetBit(1000, 2); err != nil {
t.Fatal(err)
} else if _, err := f0.ClearBit(1000, 1); err != nil {
t.Fatal(err)
@ -527,12 +527,12 @@ func BenchmarkFragment_IntersectionCount(b *testing.B) {
// Generate some intersecting data.
for i := 0; i < 10000; i += 2 {
if _, err := f.SetBit(1, uint64(i), nil, 0); err != nil {
if _, err := f.SetBit(1, uint64(i)); err != nil {
b.Fatal(err)
}
}
for i := 0; i < 10000; i += 3 {
if _, err := f.SetBit(2, uint64(i), nil, 0); err != nil {
if _, err := f.SetBit(2, uint64(i)); err != nil {
b.Fatal(err)
}
}
@ -609,7 +609,7 @@ func (f *Fragment) Reopen() error {
// This function does not accept a timestamp or quantum.
func (f *Fragment) MustSetBits(bitmapID uint64, profileIDs ...uint64) {
for _, profileID := range profileIDs {
if _, err := f.SetBit(bitmapID, profileID, nil, 0); err != nil {
if _, err := f.SetBit(bitmapID, profileID); err != nil {
panic(err)
}
}

97
frame.go
View file

@ -2,26 +2,28 @@ package pilosa
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"strconv"
"sync"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
)
const (
// FrameSuffixTime is the suffix used for time-based frames.
FrameSuffixTime = ".t"
// FrameSuffixRank is the suffix used for rank-based frames.
FrameSuffixRank = ".n"
)
// Frame represents a container for fragments.
type Frame struct {
mu sync.Mutex
path string
db string
name string
mu sync.Mutex
path string
db string
name string
timeQuantum TimeQuantum
// Fragments by slice.
fragments map[uint64]*Fragment
@ -80,6 +82,10 @@ func (f *Frame) Open() error {
return err
}
if err := f.loadMeta(); err != nil {
return err
}
if err := f.openFragments(); err != nil {
return err
}
@ -134,6 +140,45 @@ func (f *Frame) openFragments() error {
return nil
}
// loadMeta reads meta data for the frame, if any.
func (f *Frame) loadMeta() error {
var pb internal.Frame
// Read data from meta file.
buf, err := ioutil.ReadFile(filepath.Join(f.path, "meta"))
if os.IsNotExist(err) {
f.timeQuantum = ""
return nil
} else if err != nil {
return err
} else {
if err := proto.Unmarshal(buf, &pb); err != nil {
return err
}
}
// Copy metadata fields.
f.timeQuantum = TimeQuantum(pb.TimeQuantum)
return nil
}
// saveMeta writes meta data for the frame.
func (f *Frame) saveMeta() error {
// Marshal metadata.
buf, err := proto.Marshal(&internal.Frame{TimeQuantum: string(f.timeQuantum)})
if err != nil {
return err
}
// Write to meta file.
if err := ioutil.WriteFile(filepath.Join(f.path, "meta"), buf, 0666); err != nil {
return err
}
return nil
}
// Close closes the frame and its fragments.
func (f *Frame) Close() error {
f.mu.Lock()
@ -153,6 +198,34 @@ func (f *Frame) Close() error {
return nil
}
// TimeQuantum returns the time quantum for the frame.
func (f *Frame) TimeQuantum() TimeQuantum {
f.mu.Lock()
defer f.mu.Unlock()
return f.timeQuantum
}
// SetTimeQuantum sets the time quantum for the frame.
func (f *Frame) SetTimeQuantum(q TimeQuantum) error {
f.mu.Lock()
defer f.mu.Unlock()
// Validate input.
if !q.Valid() {
return ErrInvalidTimeQuantum
}
// Update value on frame.
f.timeQuantum = q
// Perist meta data to disk.
if err := f.saveMeta(); err != nil {
return err
}
return nil
}
// FragmentPath returns the path to a fragment in the frame.
func (f *Frame) FragmentPath(slice uint64) string {
return filepath.Join(f.path, strconv.FormatUint(slice, 10))
@ -213,6 +286,16 @@ func (f *Frame) newFragment(path string, slice uint64) *Fragment {
return frag
}
// SetBit sets a bit within the frame.
func (f *Frame) SetBit(bitmapID, profileID uint64) (changed bool, err error) {
slice := bitmapID / SliceWidth
frag, err := f.CreateFragmentIfNotExists(slice)
if err != nil {
return changed, err
}
return frag.SetBit(bitmapID, profileID)
}
type frameSlice []*Frame
func (p frameSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }

35
frame_test.go
View file

@ -34,6 +34,26 @@ func TestFrame_CreateFragmentIfNotExists(t *testing.T) {
}
}
// Ensure frame can set its time quantum.
func TestFrame_SetTimeQuantum(t *testing.T) {
f := MustOpenFrame()
defer f.Close()
// Set & retrieve time quantum.
if err := f.SetTimeQuantum(pilosa.TimeQuantum("YMDH")); err != nil {
t.Fatal(err)
} else if q := f.TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum: %s", q)
}
// Reload frame and verify that it is persisted.
if err := f.Reopen(); err != nil {
t.Fatal(err)
} else if q := f.TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum (reopen): %s", q)
}
}
// Frame represents a test wrapper for pilosa.Frame.
type Frame struct {
*pilosa.Frame
@ -63,3 +83,18 @@ func (f *Frame) Close() error {
defer os.RemoveAll(f.Path())
return f.Frame.Close()
}
// Reopen closes the database and reopens it.
func (f *Frame) Reopen() error {
if err := f.Frame.Close(); err != nil {
return err
}
path, db, name := f.Path(), f.DB(), f.Name()
f.Frame = pilosa.NewFrame(path, db, name)
if err := f.Open(); err != nil {
return err
}
return nil
}

184
handler.go
View file

@ -114,6 +114,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/db/time_quantum":
switch r.Method {
case "PATCH":
h.handlePatchDBTimeQuantum(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/db/attr/diff":
switch r.Method {
case "POST":
@ -121,6 +128,20 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/frame":
switch r.Method {
case "DELETE":
h.handleDeleteFrame(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/frame/time_quantum":
switch r.Method {
case "PATCH":
h.handlePatchFrameTimeQuantum(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/frame/attr/diff":
switch r.Method {
case "POST":
@ -203,7 +224,6 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
// Build execution options.
opt := &ExecOptions{
Timestamp: req.Timestamp,
Quantum: req.Quantum,
Remote: req.Remote,
}
@ -256,7 +276,7 @@ func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) erro
sm := h.Index.SliceN()
if strings.Contains(r.Header.Get("Accept"), "application/x-protobuf") {
pb := &internal.SliceMaxResponse{
SliceMax: &sm,
SliceMax: sm,
}
if buf, err := proto.Marshal(pb); err != nil {
return err
@ -299,6 +319,48 @@ type deleteDBRequest struct {
type deleteDBResponse struct{}
// handlePatchDBTimeQuantum handles PATCH /db/time_quantum request.
func (h *Handler) handlePatchDBTimeQuantum(w http.ResponseWriter, r *http.Request) {
// Decode request.
var req patchDBTimeQuantumRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Validate quantum.
tq, err := ParseTimeQuantum(req.TimeQuantum)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Retrieve database by name.
db, err := h.Index.CreateDBIfNotExists(req.DB)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
// Set default time quantum on database.
if err := db.SetTimeQuantum(tq); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
// Encode response.
if err := json.NewEncoder(w).Encode(patchDBTimeQuantumResponse{}); err != nil {
h.logger().Printf("response encoding error: %s", err)
}
}
type patchDBTimeQuantumRequest struct {
DB string `json:"db"`
TimeQuantum string `json:"time_quantum"`
}
type patchDBTimeQuantumResponse struct{}
// handlePostDBAttrDiff handles POST /db/attr/diff requests.
func (h *Handler) handlePostDBAttrDiff(w http.ResponseWriter, r *http.Request) {
// Decode request.
@ -355,6 +417,86 @@ type postDBAttrDiffResponse struct {
Attrs map[uint64]map[string]interface{} `json:"attrs"`
}
// handleDeleteFrame handles DELETE /frame request.
func (h *Handler) handleDeleteFrame(w http.ResponseWriter, r *http.Request) {
// Decode request.
var req deleteFrameRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Find database.
db := h.Index.DB(req.DB)
if db == nil {
if err := json.NewEncoder(w).Encode(deleteDBResponse{}); err != nil {
h.logger().Printf("response encoding error: %s", err)
}
return
}
// Delete frame from the database.
if err := db.DeleteFrame(req.Frame); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
// Encode response.
if err := json.NewEncoder(w).Encode(deleteFrameResponse{}); err != nil {
h.logger().Printf("response encoding error: %s", err)
}
}
type deleteFrameRequest struct {
DB string `json:"db"`
Frame string `json:"frame"`
}
type deleteFrameResponse struct{}
// handlePatchFrameTimeQuantum handles PATCH /frame/time_quantum request.
func (h *Handler) handlePatchFrameTimeQuantum(w http.ResponseWriter, r *http.Request) {
// Decode request.
var req patchFrameTimeQuantumRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Validate quantum.
tq, err := ParseTimeQuantum(req.TimeQuantum)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Retrieve database by name.
f, err := h.Index.CreateFrameIfNotExists(req.DB, req.Frame)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
// Set default time quantum on database.
if err := f.SetTimeQuantum(tq); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
// Encode response.
if err := json.NewEncoder(w).Encode(patchFrameTimeQuantumResponse{}); err != nil {
h.logger().Printf("response encoding error: %s", err)
}
}
type patchFrameTimeQuantumRequest struct {
DB string `json:"db"`
Frame string `json:"frame"`
TimeQuantum string `json:"time_quantum"`
}
type patchFrameTimeQuantumResponse struct{}
// handlePostFrameAttrDiff handles POST /frame/attr/diff requests.
func (h *Handler) handlePostFrameAttrDiff(w http.ResponseWriter, r *http.Request) {
// Decode request.
@ -495,7 +637,7 @@ func (h *Handler) readURLQueryRequest(r *http.Request) (*QueryRequest, error) {
}
// Parse time granularity.
quantum := YMDH
quantum := TimeQuantum("YMDH")
if s := q.Get("time_granularity"); s != "" {
v, err := ParseTimeQuantum(s)
if err != nil {
@ -561,7 +703,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
db, frame, slice := req.GetDB(), req.GetFrame(), req.GetSlice()
db, frame, slice := req.DB, req.Frame, req.Slice
// Validate that this handler owns the slice.
if !h.Cluster.OwnsFragment(h.Host, db, slice) {
@ -578,16 +720,16 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
http.Error(w, "fragment error", http.StatusInternalServerError)
return
}
h.logger().Println("Import into Fragment:", db, frame, slice, len(req.GetProfileIDs()))
h.logger().Println("Import into Fragment:", db, frame, slice, len(req.ProfileIDs))
// Import into fragment.
err = f.Import(req.GetBitmapIDs(), req.GetProfileIDs())
err = f.Import(req.BitmapIDs, req.ProfileIDs)
if err != nil {
h.logger().Printf("import error: db=%s, frame=%s, slice=%d, bits=%d, err=%s", db, frame, slice, len(req.GetProfileIDs()), err)
h.logger().Printf("import error: db=%s, frame=%s, slice=%d, bits=%d, err=%s", db, frame, slice, len(req.ProfileIDs), err)
}
// Marshal response object.
buf, e := proto.Marshal(&internal.ImportResponse{Err: proto.String(errorString(err))})
buf, e := proto.Marshal(&internal.ImportResponse{Err: errorString(err)})
if e != nil {
http.Error(w, fmt.Sprintf("marshal import response: %s", err), http.StatusInternalServerError)
return
@ -733,7 +875,7 @@ func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Requ
}
// Retrieve fragment from index.
f := h.Index.Fragment(req.GetDB(), req.GetFrame(), req.GetSlice())
f := h.Index.Fragment(req.DB, req.Frame, req.Slice)
if f == nil {
http.Error(w, ErrFragmentNotFound.Error(), http.StatusNotFound)
return
@ -742,7 +884,7 @@ func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Requ
// Read data
var resp internal.BlockDataResponse
if f != nil {
resp.BitmapIDs, resp.ProfileIDs = f.BlockData(int(req.GetBlock()))
resp.BitmapIDs, resp.ProfileIDs = f.BlockData(int(req.Block))
}
// Encode response.
@ -919,16 +1061,16 @@ type QueryRequest struct {
func decodeQueryRequest(pb *internal.QueryRequest) *QueryRequest {
req := &QueryRequest{
DB: pb.GetDB(),
Query: pb.GetQuery(),
Slices: pb.GetSlices(),
Profiles: pb.GetProfiles(),
Quantum: TimeQuantum(pb.GetQuantum()),
Remote: pb.GetRemote(),
DB: pb.DB,
Query: pb.Query,
Slices: pb.Slices,
Profiles: pb.Profiles,
Quantum: TimeQuantum(pb.Quantum),
Remote: pb.Remote,
}
if pb.Timestamp != nil {
t := time.Unix(0, pb.GetTimestamp())
if pb.Timestamp != 0 {
t := time.Unix(0, pb.Timestamp)
req.Timestamp = &t
}
@ -978,14 +1120,14 @@ func encodeQueryResponse(resp *QueryResponse) *internal.QueryResponse {
case []Pair:
pb.Results[i].Pairs = encodePairs(result)
case uint64:
pb.Results[i].N = proto.Uint64(result)
pb.Results[i].N = result
case bool:
pb.Results[i].Changed = proto.Bool(result)
pb.Results[i].Changed = result
}
}
if resp.Err != nil {
pb.Err = proto.String(resp.Err.Error())
pb.Err = resp.Err.Error()
}
return pb

101
handler_test.go
View file

@ -93,8 +93,8 @@ func TestHandler_Query_Args_Protobuf(t *testing.T) {
// Generate request body.
reqBody, err := proto.Marshal(&internal.QueryRequest{
DB: proto.String("db0"),
Query: proto.String("Count(Bitmap(100))"),
DB: "db0",
Query: "Count(Bitmap(100))",
Slices: []uint64{0, 1},
})
if err != nil {
@ -157,7 +157,7 @@ func TestHandler_Query_Uint64_Protobuf(t *testing.T) {
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if n := resp.Results[0].GetN(); n != 100 {
} else if n := resp.Results[0].N; n != 100 {
t.Fatalf("unexpected n: %d", n)
}
}
@ -232,15 +232,15 @@ func TestHandler_Query_Bitmap_Protobuf(t *testing.T) {
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if bits := resp.Results[0].GetBitmap().GetBits(); !reflect.DeepEqual(bits, []uint64{1, SliceWidth + 1}) {
} else if bits := resp.Results[0].Bitmap.Bits; !reflect.DeepEqual(bits, []uint64{1, SliceWidth + 1}) {
t.Fatalf("unexpected bits: %+v", bits)
} else if attrs := resp.Results[0].GetBitmap().GetAttrs(); len(attrs) != 3 {
} else if attrs := resp.Results[0].Bitmap.Attrs; len(attrs) != 3 {
t.Fatalf("unexpected attr length: %d", len(attrs))
} else if k, v := attrs[0].GetKey(), attrs[0].GetStringValue(); k != "a" || v != "b" {
} else if k, v := attrs[0].Key, attrs[0].StringValue; k != "a" || v != "b" {
t.Fatalf("unexpected attr[0]: %s=%v", k, v)
} else if k, v := attrs[1].GetKey(), attrs[1].GetUintValue(); k != "c" || v != uint64(1) {
} else if k, v := attrs[1].Key, attrs[1].UintValue; k != "c" || v != uint64(1) {
t.Fatalf("unexpected attr[1]: %s=%v", k, v)
} else if k, v := attrs[2].GetKey(), attrs[2].GetBoolValue(); k != "d" || v != true {
} else if k, v := attrs[2].Key, attrs[2].BoolValue; k != "d" || v != true {
t.Fatalf("unexpected attr[2]: %s=%v", k, v)
}
}
@ -268,9 +268,9 @@ func TestHandler_Query_Bitmap_Profiles_Protobuf(t *testing.T) {
// Encode request body.
buf, err := proto.Marshal(&internal.QueryRequest{
DB: proto.String("d"),
Query: proto.String("Bitmap(100)"),
Profiles: proto.Bool(true),
DB: "d",
Query: "Bitmap(100)",
Profiles: true,
})
if err != nil {
t.Fatal(err)
@ -289,25 +289,25 @@ func TestHandler_Query_Bitmap_Profiles_Protobuf(t *testing.T) {
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if bits := resp.Results[0].GetBitmap().GetBits(); !reflect.DeepEqual(bits, []uint64{1, SliceWidth + 1}) {
if bits := resp.Results[0].Bitmap.Bits; !reflect.DeepEqual(bits, []uint64{1, SliceWidth + 1}) {
t.Fatalf("unexpected bits: %+v", bits)
} else if attrs := resp.Results[0].GetBitmap().GetAttrs(); len(attrs) != 3 {
} else if attrs := resp.Results[0].Bitmap.Attrs; len(attrs) != 3 {
t.Fatalf("unexpected attr length: %d", len(attrs))
} else if k, v := attrs[0].GetKey(), attrs[0].GetStringValue(); k != "a" || v != "b" {
} else if k, v := attrs[0].Key, attrs[0].StringValue; k != "a" || v != "b" {
t.Fatalf("unexpected attr[0]: %s=%v", k, v)
} else if k, v := attrs[1].GetKey(), attrs[1].GetUintValue(); k != "c" || v != uint64(1) {
} else if k, v := attrs[1].Key, attrs[1].UintValue; k != "c" || v != uint64(1) {
t.Fatalf("unexpected attr[1]: %s=%v", k, v)
} else if k, v := attrs[2].GetKey(), attrs[2].GetBoolValue(); k != "d" || v != true {
} else if k, v := attrs[2].Key, attrs[2].BoolValue; k != "d" || v != true {
t.Fatalf("unexpected attr[2]: %s=%v", k, v)
}
if a := resp.GetProfiles(); len(a) != 1 {
if a := resp.Profiles; len(a) != 1 {
t.Fatalf("unexpected profiles length: %d", len(a))
} else if a[0].GetID() != 1 {
t.Fatalf("unexpected id: %d", a[0].GetID())
} else if len(a[0].GetAttrs()) != 1 {
} else if a[0].ID != 1 {
t.Fatalf("unexpected id: %d", a[0].ID)
} else if len(a[0].Attrs) != 1 {
t.Fatalf("unexpected profile attr length: %d", len(a))
} else if k, v := a[0].GetAttrs()[0].GetKey(), a[0].GetAttrs()[0].GetStringValue(); k != "x" || v != "y" {
} else if k, v := a[0].Attrs[0].Key, a[0].Attrs[0].StringValue; k != "x" || v != "y" {
t.Fatalf("unexpected attr[0]: %s=%v", k, v)
}
}
@ -391,7 +391,7 @@ func TestHandler_Query_Err_Protobuf(t *testing.T) {
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if s := resp.GetErr(); s != `marker` {
} else if s := resp.Err; s != `marker` {
t.Fatalf("unexpected error: %s", s)
}
}
@ -453,6 +453,63 @@ func TestHandler_DB_Delete(t *testing.T) {
}
}
// Ensure handler can delete a frame.
func TestHandler_DeleteFrame(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
if _, err := idx.CreateFrameIfNotExists("d0", "f1"); err != nil {
t.Fatal(err)
}
h := NewHandler()
h.Index = idx.Index
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("DELETE", "/frame", strings.NewReader(`{"db":"d0","frame":"f1"}`)))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{}`+"\n" {
t.Fatalf("unexpected body: %s", body)
} else if f := idx.DB("d0").Frame("f1"); f != nil {
t.Fatal("expected nil frame")
}
}
// Ensure handler can set the DB time quantum.
func TestHandler_SetDBTimeQuantum(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
h := NewHandler()
h.Index = idx.Index
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/time_quantum", strings.NewReader(`{"db":"d0","time_quantum":"ymdh"}`)))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{}`+"\n" {
t.Fatalf("unexpected body: %s", body)
} else if q := idx.DB("d0").TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected time quantum: %s", q)
}
}
// Ensure handler can set the frame time quantum.
func TestHandler_SetFrameTimeQuantum(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
h := NewHandler()
h.Index = idx.Index
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/frame/time_quantum", strings.NewReader(`{"db":"d0","frame":"f1","time_quantum":"ymdh"}`)))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{}`+"\n" {
t.Fatalf("unexpected body: %s", body)
} else if q := idx.DB("d0").Frame("f1").TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected time quantum: %s", q)
}
}
// Ensure the handler can return data in differing blocks for a database.
func TestHandler_DB_AttrStore_Diff(t *testing.T) {
idx := MustOpenIndex()

44
index_test.go
View file

@ -19,11 +19,11 @@ func TestIndex_DeleteDB(t *testing.T) {
// Write bits to separate databases.
f0 := idx.MustCreateFragmentIfNotExists("d0", "f", 0)
if _, err := f0.SetBit(100, 200, nil, 0); err != nil {
if _, err := f0.SetBit(100, 200); err != nil {
t.Fatal(err)
}
f1 := idx.MustCreateFragmentIfNotExists("d1", "f", 0)
if _, err := f1.SetBit(100, 200, nil, 0); err != nil {
if _, err := f1.SetBit(100, 200); err != nil {
t.Fatal(err)
}
@ -74,18 +74,18 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
// Set data on the local index.
f := idx0.MustCreateFragmentIfNotExists("d", "f", 0)
if _, err := f.SetBit(0, 10, nil, 0); err != nil {
if _, err := f.SetBit(0, 10); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(2, 20, nil, 0); err != nil {
} else if _, err := f.SetBit(2, 20); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(120, 10, nil, 0); err != nil {
} else if _, err := f.SetBit(120, 10); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(200, 4, nil, 0); err != nil {
} else if _, err := f.SetBit(200, 4); err != nil {
t.Fatal(err)
}
f = idx0.MustCreateFragmentIfNotExists("d", "f0", 1)
if _, err := f.SetBit(9, SliceWidth+5, nil, 0); err != nil {
if _, err := f.SetBit(9, SliceWidth+5); err != nil {
t.Fatal(err)
}
@ -93,20 +93,20 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
// Set data on the remote index.
f = idx1.MustCreateFragmentIfNotExists("d", "f", 0)
if _, err := f.SetBit(0, 4000, nil, 0); err != nil {
if _, err := f.SetBit(0, 4000); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(3, 10, nil, 0); err != nil {
} else if _, err := f.SetBit(3, 10); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(120, 10, nil, 0); err != nil {
} else if _, err := f.SetBit(120, 10); err != nil {
t.Fatal(err)
}
f = idx1.MustCreateFragmentIfNotExists("y", "z", 3)
if _, err := f.SetBit(10, (3*SliceWidth)+4, nil, 0); err != nil {
if _, err := f.SetBit(10, (3*SliceWidth)+4); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(10, (3*SliceWidth)+5, nil, 0); err != nil {
} else if _, err := f.SetBit(10, (3*SliceWidth)+5); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(10, (3*SliceWidth)+7, nil, 0); err != nil {
} else if _, err := f.SetBit(10, (3*SliceWidth)+7); err != nil {
t.Fatal(err)
}
@ -185,6 +185,24 @@ func (i *Index) Close() error {
return i.Index.Close()
}
// MustCreateDBIfNotExists returns a given db. Panic on error.
func (i *Index) MustCreateDBIfNotExists(db string) *DB {
d, err := i.Index.CreateDBIfNotExists(db)
if err != nil {
panic(err)
}
return &DB{DB: d}
}
// MustCreateFrameIfNotExists returns a given frame. Panic on error.
func (i *Index) MustCreateFrameIfNotExists(db, frame string) *Frame {
f, err := i.Index.CreateFrameIfNotExists(db, frame)
if err != nil {
panic(err)
}
return &Frame{Frame: f}
}
// MustCreateFragmentIfNotExists returns a given fragment. Panic on error.
func (i *Index) MustCreateFragmentIfNotExists(db, frame string, slice uint64) *Fragment {
f, err := i.Index.CreateFragmentIfNotExists(db, frame, slice)

2
internal/internal.go
View file

@ -7,6 +7,8 @@ import (
"github.com/gogo/protobuf/proto"
)
//go:generate protoc --gofast_out=. internal.proto
type Request proto.Message
type Response proto.Message

3847
internal/internal.pb.go
View file

File diff suppressed because it is too large Load diff

84
internal/internal.proto
View file

@ -1,30 +1,41 @@
syntax = "proto3";
package internal;
message DB {
string TimeQuantum = 1;
}
message Frame {
string TimeQuantum = 1;
}
message Bitmap {
repeated uint64 Bits = 1;
repeated Attr Attrs = 2;
repeated uint64 Bits = 1;
repeated Attr Attrs = 2;
}
message Pair {
required uint64 Key = 1;
required uint64 Count = 2;
uint64 Key = 1;
uint64 Count = 2;
}
message Bit {
required uint64 BitmapID = 1;
required uint64 ProfileID = 2;
uint64 BitmapID = 1;
uint64 ProfileID = 2;
}
message Profile {
required uint64 ID = 1;
uint64 ID = 1;
repeated Attr Attrs = 2;
}
message Attr {
required string Key = 1;
optional string StringValue = 2;
optional uint64 UintValue = 3;
optional bool BoolValue = 4;
string Key = 1;
uint64 Type = 2;
string StringValue = 3;
uint64 UintValue = 4;
bool BoolValue = 5;
}
message AttrMap {
@ -32,49 +43,49 @@ message AttrMap {
}
message QueryRequest {
required string DB = 1;
required string Query = 2;
repeated uint64 Slices = 3;
optional bool Profiles = 4;
optional int64 Timestamp = 5;
optional uint32 Quantum = 6;
optional bool Remote = 7;
string DB = 1;
string Query = 2;
repeated uint64 Slices = 3;
bool Profiles = 4;
int64 Timestamp = 5;
string Quantum = 6;
bool Remote = 7;
}
message QueryResponse {
optional string Err = 1;
repeated QueryResult Results = 2;
repeated Profile Profiles = 3;
string Err = 1;
repeated QueryResult Results = 2;
repeated Profile Profiles = 3;
}
message QueryResult {
optional Bitmap Bitmap = 1;
optional uint64 N = 2;
repeated Pair Pairs = 3;
optional bool Changed = 4;
Bitmap Bitmap = 1;
uint64 N = 2;
repeated Pair Pairs = 3;
bool Changed = 4;
}
message ImportRequest {
required string DB = 1;
required string Frame = 2;
required uint64 Slice = 3;
repeated uint64 BitmapIDs = 4;
string DB = 1;
string Frame = 2;
uint64 Slice = 3;
repeated uint64 BitmapIDs = 4;
repeated uint64 ProfileIDs = 5;
}
message ImportResponse {
optional string Err = 1;
string Err = 1;
}
message BlockDataRequest {
required string DB = 1;
required string Frame = 2;
required uint64 Slice = 3;
required uint64 Block = 4;
string DB = 1;
string Frame = 2;
uint64 Slice = 3;
uint64 Block = 4;
}
message BlockDataResponse {
repeated uint64 BitmapIDs = 1;
repeated uint64 BitmapIDs = 1;
repeated uint64 ProfileIDs = 2;
}
@ -83,5 +94,6 @@ message Cache {
}
message SliceMaxResponse {
required uint64 SliceMax = 1;
uint64 SliceMax = 1;
}

13
pilosa.go
View file

@ -3,12 +3,9 @@ package pilosa
import (
"errors"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
)
//go:generate protoc --gogo_out=. internal/internal.proto
var (
// ErrHostRequired is returned when excuting a remote operation without a host.
ErrHostRequired = errors.New("host required")
@ -57,7 +54,7 @@ func decodeProfiles(a []*internal.Profile) []*Profile {
// encodeProfile converts p into its internal representation.
func encodeProfile(p *Profile) *internal.Profile {
return &internal.Profile{
ID: proto.Uint64(p.ID),
ID: p.ID,
Attrs: encodeAttrs(p.Attrs),
}
}
@ -65,12 +62,12 @@ func encodeProfile(p *Profile) *internal.Profile {
// decodeProfile converts b from its internal representation.
func decodeProfile(pb *internal.Profile) *Profile {
p := &Profile{
ID: pb.GetID(),
ID: pb.ID,
}
if len(pb.GetAttrs()) > 0 {
p.Attrs = make(map[string]interface{}, len(pb.GetAttrs()))
for _, attr := range pb.GetAttrs() {
if len(pb.Attrs) > 0 {
p.Attrs = make(map[string]interface{}, len(pb.Attrs))
for _, attr := range pb.Attrs {
k, v := decodeAttr(attr)
p.Attrs[k] = v
}

3
server.go
View file

@ -255,6 +255,5 @@ func checkMaxSlice(hostport string) (uint64, error) {
return 0, err
}
return *pb.SliceMax, nil
return pb.SliceMax, nil
}

228
time.go
View file

@ -2,122 +2,166 @@ package pilosa
import (
"errors"
"fmt"
"strings"
"time"
)
// TimeQuantum represents a time granularity for time-based bitmap ids.
type TimeQuantum uint
// ErrInvalidTimeQuantum is returned when parsing a time quantum.
var ErrInvalidTimeQuantum = errors.New("invalid time quantum")
// ParseTimeQuantum parses s into a quantum.
func ParseTimeQuantum(s string) (TimeQuantum, error) {
switch strings.ToUpper(s) {
case "Y":
return Y, nil
case "M":
return YM, nil
case "D":
return YMD, nil
case "H":
return YMDH, nil
default:
return 0, errors.New("invalid quantum")
}
}
// TimeQuantum represents a time granularity for time-based bitmaps.
type TimeQuantum string
// String returns the string representation of the quantum.
func (q TimeQuantum) String() string {
// HasYear returns true if the quantum contains a 'Y' unit.
func (q TimeQuantum) HasYear() bool { return strings.ContainsRune(string(q), 'Y') }
// HasMonth returns true if the quantum contains a 'M' unit.
func (q TimeQuantum) HasMonth() bool { return strings.ContainsRune(string(q), 'M') }
// HasDay returns true if the quantum contains a 'D' unit.
func (q TimeQuantum) HasDay() bool { return strings.ContainsRune(string(q), 'D') }
// HasHour returns true if the quantum contains a 'H' unit.
func (q TimeQuantum) HasHour() bool { return strings.ContainsRune(string(q), 'H') }
// Valid returns true if q is a valid time quantum value.
func (q TimeQuantum) Valid() bool {
switch q {
case Y:
return "Y"
case YM:
return "M"
case YMD:
return "D"
case YMDH:
return "H"
case "Y", "YM", "YMD", "YMDH",
"M", "MD", "MDH",
"D", "DH",
"H",
"":
return true
default:
return "Y"
return false
}
}
const (
Y TimeQuantum = 3
YM TimeQuantum = 2
YMD TimeQuantum = 1
YMDH TimeQuantum = 0
)
// TimeID returns a packed time identifier from a year/month/day/hour & tile.
func TimeID(q TimeQuantum, year uint, month uint, day uint, hour uint, tileID uint64) uint64 {
v := uint64((uint(q) << 30) | ((year - 1970) << 23) | (month << 19) | (day << 14) | (hour << 9))
return (v << 32) | tileID
// ParseTimeQuantum parses v into a time quantum.
func ParseTimeQuantum(v string) (TimeQuantum, error) {
q := TimeQuantum(strings.ToUpper(v))
if !q.Valid() {
return "", ErrInvalidTimeQuantum
}
return q, nil
}
// TimeIDsFromRange returns timestamp bitmap ids within a time range.
func TimeIDsFromRange(start, end time.Time, tileID uint64) []uint64 {
// FrameByTimeUnit returns the frame name for time with a given quantum unit.
func FrameByTimeUnit(name string, t time.Time, unit rune) string {
switch unit {
case 'Y':
return fmt.Sprintf("%s_%s", name, t.Format("2006"))
case 'M':
return fmt.Sprintf("%s_%s", name, t.Format("200601"))
case 'D':
return fmt.Sprintf("%s_%s", name, t.Format("20060102"))
case 'H':
return fmt.Sprintf("%s_%s", name, t.Format("2006010215"))
default:
return ""
}
}
// FramesByTime returns a list of frames for a given timestamp.
func FramesByTime(name string, t time.Time, q TimeQuantum) []string {
a := make([]string, 0, len(q))
for _, unit := range q {
frame := FrameByTimeUnit(name, t, unit)
if frame == "" {
continue
}
a = append(a, frame)
}
return a
}
// FramesByTimeRange returns a list of frames to traverse to query a time range.
func FramesByTimeRange(name string, start, end time.Time, q TimeQuantum) []string {
t := start
var results []uint64
for t.Before(end) {
if !nextDay(t, end) {
// Save flags for performance.
hasYear := q.HasYear()
hasMonth := q.HasMonth()
hasDay := q.HasDay()
hasHour := q.HasHour()
var results []string
// Walk up from smallest units to largest units.
if hasHour || hasDay || hasMonth {
for t.Before(end) {
if hasHour {
if !nextDayGTE(t, end) {
break
} else if t.Hour() != 0 {
results = append(results, FrameByTimeUnit(name, t, 'H'))
t = t.Add(time.Hour)
continue
}
}
if hasDay {
if !nextMonthGTE(t, end) {
break
} else if t.Day() != 1 {
results = append(results, FrameByTimeUnit(name, t, 'D'))
t = t.AddDate(0, 0, 1)
continue
}
}
if hasMonth {
if !nextYearGTE(t, end) {
break
} else if t.Month() != 1 {
results = append(results, FrameByTimeUnit(name, t, 'M'))
t = t.AddDate(0, 1, 0)
continue
}
}
// If a unit exists but isn't set and there are no larger units
// available then we need to exit the loop because we are no longer
// making progress.
break
}
if t.Hour() == 0 {
if !nextMonth(t, end) {
break
}
if t.Day() == 1 {
if !nextYear(t, end) {
break
}
if t.Month() == 1 {
break
}
results = append(results, TimeID(YM, uint(t.Year()), uint(t.Month()), 0, 0, tileID))
t = t.AddDate(0, 1, 0)
} else {
results = append(results, TimeID(YMD, uint(t.Year()), uint(t.Month()), uint(t.Day()), 0, tileID))
t = t.AddDate(0, 0, 1)
}
} else {
results = append(results, TimeID(YMDH, uint(t.Year()), uint(t.Month()), uint(t.Day()), uint(t.Hour()), tileID))
t = t.Add(time.Hour)
}
}
// Walk back down from largest units to smallest units.
for t.Before(end) {
if nextYear(t, end) {
results = append(results, TimeID(Y, uint(t.Year()), 0, 0, 0, tileID))
if hasYear && nextYearGTE(t, end) {
results = append(results, FrameByTimeUnit(name, t, 'Y'))
t = t.AddDate(1, 0, 0)
} else if nextMonth(t, end) {
results = append(results, TimeID(YM, uint(t.Year()), uint(t.Month()), 0, 0, tileID))
} else if hasMonth && nextMonthGTE(t, end) {
results = append(results, FrameByTimeUnit(name, t, 'M'))
t = t.AddDate(0, 1, 0)
} else if nextDay(t, end) {
results = append(results, TimeID(YMD, uint(t.Year()), uint(t.Month()), uint(t.Day()), 0, tileID))
} else if hasDay && nextDayGTE(t, end) {
results = append(results, FrameByTimeUnit(name, t, 'D'))
t = t.AddDate(0, 0, 1)
} else {
results = append(results, TimeID(YMDH, uint(t.Year()), uint(t.Month()), uint(t.Day()), uint(t.Hour()), tileID))
} else if hasHour {
results = append(results, FrameByTimeUnit(name, t, 'H'))
t = t.Add(time.Hour)
} else {
break
}
}
return results
}
func nextYear(start time.Time, end time.Time) bool {
next := start.AddDate(1, 0, 0)
func nextYearGTE(t time.Time, end time.Time) bool {
next := t.AddDate(1, 0, 0)
if next.Year() == end.Year() {
return true
}
return end.After(next)
}
func nextMonth(start time.Time, end time.Time) bool {
next := start.AddDate(0, 1, 0)
func nextMonthGTE(t time.Time, end time.Time) bool {
next := t.AddDate(0, 1, 0)
y1, m1, _ := next.Date()
y2, m2, _ := end.Date()
if (y1 == y2) && (m1 == m2) {
@ -126,8 +170,8 @@ func nextMonth(start time.Time, end time.Time) bool {
return end.After(next)
}
func nextDay(start time.Time, end time.Time) bool {
next := start.AddDate(0, 0, 1)
func nextDayGTE(t time.Time, end time.Time) bool {
next := t.AddDate(0, 0, 1)
y1, m1, d1 := next.Date()
y2, m2, d2 := end.Date()
if (y1 == y2) && (m1 == m2) && (d1 == d2) {
@ -135,23 +179,3 @@ func nextDay(start time.Time, end time.Time) bool {
}
return end.After(next)
}
// TimeIDsFromQuantum returns a list of time identifiers for a single quantum.
func TimeIDsFromQuantum(q TimeQuantum, t time.Time, tileID uint64) []uint64 {
y, m, d, h := uint(t.Year()), uint(t.Month()), uint(t.Day()), uint(t.Hour())
v := make([]uint64, 0, 4)
if q <= Y {
v = append(v, TimeID(Y, y, 0, 0, 0, tileID))
}
if q <= YM {
v = append(v, TimeID(YM, y, m, 0, 0, tileID))
}
if q <= YMD {
v = append(v, TimeID(YMD, y, m, d, 0, tileID))
}
if q <= YMDH {
v = append(v, TimeID(YMDH, y, m, d, h, tileID))
}
return v
}

242
time_test.go
View file

@ -1,134 +1,162 @@
package pilosa_test
import (
"reflect"
"testing"
"time"
"github.com/pilosa/pilosa"
)
func TestTimeIDsFromRange_1h_0(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-08-11 14:00"),
*MustParseTime("2014-08-11 16:00"),
uint64(1),
); len(m) != 2 {
t.Fatalf("unexpected range len: %d", len(m))
}
// Ensure string can be parsed into time quantum.
func TestParseTimeQuantum(t *testing.T) {
t.Run("OK", func(t *testing.T) {
if q, err := pilosa.ParseTimeQuantum("YMDH"); err != nil {
t.Fatalf("unexpected error: %s", err)
} else if q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum: %#v", q)
}
})
t.Run("ErrInvalidTimeQuantum", func(t *testing.T) {
if _, err := pilosa.ParseTimeQuantum("BADQUANTUM"); err != pilosa.ErrInvalidTimeQuantum {
t.Fatalf("unexpected error: %s", err)
}
})
}
func TestTimeIDsFromRange_1h_1(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-02 10:03"),
*MustParseTime("2014-01-02 11:03"),
uint64(1),
); len(m) != 1 {
t.Fatalf("unexpected range len: %d", len(m))
}
// Ensure generated frame name can be returned for a given time unit.
func TestFrameByTimeUnit(t *testing.T) {
ts := time.Date(2000, time.January, 2, 3, 4, 5, 6, time.UTC)
t.Run("Y", func(t *testing.T) {
if s := pilosa.FrameByTimeUnit("F", ts, 'Y'); s != "F_2000" {
t.Fatalf("unexpected name: %s", s)
}
})
t.Run("M", func(t *testing.T) {
if s := pilosa.FrameByTimeUnit("F", ts, 'M'); s != "F_200001" {
t.Fatalf("unexpected name: %s", s)
}
})
t.Run("D", func(t *testing.T) {
if s := pilosa.FrameByTimeUnit("F", ts, 'D'); s != "F_20000102" {
t.Fatalf("unexpected name: %s", s)
}
})
t.Run("H", func(t *testing.T) {
if s := pilosa.FrameByTimeUnit("F", ts, 'H'); s != "F_2000010203" {
t.Fatalf("unexpected name: %s", s)
}
})
}
func TestTimeIDsFromRange_2h(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-02 10:03"),
*MustParseTime("2014-01-02 12:03"),
uint64(1),
); len(m) != 2 {
t.Fatalf("unexpected range len: %d", len(m))
}
// Ensure all applicable frame names can be generated when mutating a time bit.
func TestFramesByTime(t *testing.T) {
ts := time.Date(2000, time.January, 2, 3, 4, 5, 6, time.UTC)
t.Run("YMDH", func(t *testing.T) {
a := pilosa.FramesByTime("F", ts, MustParseTimeQuantum("YMDH"))
if !reflect.DeepEqual(a, []string{"F_2000", "F_200001", "F_20000102", "F_2000010203"}) {
t.Fatalf("unexpected names: %+v", a)
}
})
t.Run("D", func(t *testing.T) {
a := pilosa.FramesByTime("F", ts, MustParseTimeQuantum("D"))
if !reflect.DeepEqual(a, []string{"F_20000102"}) {
t.Fatalf("unexpected names: %+v", a)
}
})
}
func TestTimeIDsFromRange_24h(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-02 12:03"),
*MustParseTime("2014-01-03 12:03"),
uint64(1),
); len(m) != 24 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromRange_1d(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-02 00:00"),
*MustParseTime("2014-01-03 00:00"),
uint64(1),
); len(m) != 1 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromRange_1d1h(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-02 00:00"),
*MustParseTime("2014-01-03 01:00"),
uint64(1),
); len(m) != 2 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromRange_1h1d(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-02 23:00"),
*MustParseTime("2014-01-04 00:00"),
uint64(1),
); len(m) != 2 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromRange_1h1d1h(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-02 23:00"),
*MustParseTime("2014-01-04 01:00"),
uint64(1),
); len(m) != 3 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromRange_1y(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-01 00:00"),
*MustParseTime("2015-01-01 00:00"),
uint64(1),
); len(m) != 1 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromRange_1h1d1m(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-30 23:00"),
*MustParseTime("2014-03-01 00:00"),
uint64(1),
); len(m) != 3 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromRange_1h1d1m1d1h(t *testing.T) {
if m := pilosa.TimeIDsFromRange(
*MustParseTime("2014-01-30 23:00"),
*MustParseTime("2014-03-02 01:00"),
uint64(1),
); len(m) != 5 {
t.Fatalf("unexpected range len: %d", len(m))
}
}
func TestTimeIDsFromQuantum(t *testing.T) {
_ = pilosa.TimeIDsFromQuantum(pilosa.YMD, *MustParseTime("1970-01-01 00:00"), uint64(15027))
// Ensure sets of frames can be returned for a given time range.
func TestFramesByTimeRange(t *testing.T) {
t.Run("Y", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2002-01-01 00:00"), MustParseTimeQuantum("Y"))
if !reflect.DeepEqual(a, []string{"F_2000", "F_2001"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("YM", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-11-01 00:00"), MustParseTime("2003-03-01 00:00"), MustParseTimeQuantum("YM"))
if !reflect.DeepEqual(a, []string{"F_200011", "F_200012", "F_2001", "F_2002", "F_200301", "F_200302"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("YMD", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-11-28 00:00"), MustParseTime("2003-03-02 00:00"), MustParseTimeQuantum("YMD"))
if !reflect.DeepEqual(a, []string{"F_20001128", "F_20001129", "F_20001130", "F_200012", "F_2001", "F_2002", "F_200301", "F_200302", "F_20030301"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("YMDH", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-11-28 22:00"), MustParseTime("2002-03-01 03:00"), MustParseTimeQuantum("YMDH"))
if !reflect.DeepEqual(a, []string{"F_2000112822", "F_2000112823", "F_20001129", "F_20001130", "F_200012", "F_2001", "F_200201", "F_200202", "F_2002030100", "F_2002030101", "F_2002030102"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("M", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2000-03-01 00:00"), MustParseTimeQuantum("M"))
if !reflect.DeepEqual(a, []string{"F_200001", "F_200002"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("MD", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-11-29 00:00"), MustParseTime("2002-02-03 00:00"), MustParseTimeQuantum("MD"))
if !reflect.DeepEqual(a, []string{"F_20001129", "F_20001130", "F_200012", "F_200101", "F_200102", "F_200103", "F_200104", "F_200105", "F_200106", "F_200107", "F_200108", "F_200109", "F_200110", "F_200111", "F_200112", "F_200201", "F_20020201", "F_20020202"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("MDH", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-11-29 22:00"), MustParseTime("2002-03-02 03:00"), MustParseTimeQuantum("MDH"))
if !reflect.DeepEqual(a, []string{"F_2000112922", "F_2000112923", "F_20001130", "F_200012", "F_200101", "F_200102", "F_200103", "F_200104", "F_200105", "F_200106", "F_200107", "F_200108", "F_200109", "F_200110", "F_200111", "F_200112", "F_200201", "F_200202", "F_20020301", "F_2002030200", "F_2002030201", "F_2002030202"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("D", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2000-01-04 00:00"), MustParseTimeQuantum("D"))
if !reflect.DeepEqual(a, []string{"F_20000101", "F_20000102", "F_20000103"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("DH", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-01-01 22:00"), MustParseTime("2000-03-01 02:00"), MustParseTimeQuantum("DH"))
if !reflect.DeepEqual(a, []string{"F_2000010122", "F_2000010123", "F_20000102", "F_20000103", "F_20000104", "F_20000105", "F_20000106", "F_20000107", "F_20000108", "F_20000109", "F_20000110", "F_20000111", "F_20000112", "F_20000113", "F_20000114", "F_20000115", "F_20000116", "F_20000117", "F_20000118", "F_20000119", "F_20000120", "F_20000121", "F_20000122", "F_20000123", "F_20000124", "F_20000125", "F_20000126", "F_20000127", "F_20000128", "F_20000129", "F_20000130", "F_20000131", "F_20000201", "F_20000202", "F_20000203", "F_20000204", "F_20000205", "F_20000206", "F_20000207", "F_20000208", "F_20000209", "F_20000210", "F_20000211", "F_20000212", "F_20000213", "F_20000214", "F_20000215", "F_20000216", "F_20000217", "F_20000218", "F_20000219", "F_20000220", "F_20000221", "F_20000222", "F_20000223", "F_20000224", "F_20000225", "F_20000226", "F_20000227", "F_20000228", "F_20000229", "F_2000030100", "F_2000030101"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
t.Run("H", func(t *testing.T) {
a := pilosa.FramesByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2000-01-01 02:00"), MustParseTimeQuantum("H"))
if !reflect.DeepEqual(a, []string{"F_2000010100", "F_2000010101"}) {
t.Fatalf("unexpected frames: %#v", a)
}
})
}
// DefaultTimeLayout is the time layout used by the tests.
const DefaultTimeLayout = "2006-01-02 15:04"
// MustParseTime parses value using DefaultTimeLayout. Panic on error.
func MustParseTime(value string) *time.Time {
func MustParseTime(value string) time.Time {
v, err := time.Parse(DefaultTimeLayout, value)
if err != nil {
panic(err)
}
return v
}
// MustParseTimePtr parses value using DefaultTimeLayout. Panic on error.
func MustParseTimePtr(value string) *time.Time {
v := MustParseTime(value)
return &v
}
// MustParseTimeQuantum parses v into a time quantum. Panic on error.
func MustParseTimeQuantum(v string) pilosa.TimeQuantum {
q, err := pilosa.ParseTimeQuantum(v)
if err != nil {
panic(err)
}
return q
}