scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into benchmark-runner

Browse source
This commit is contained in:
jaffee 2017-01-09 16:55:03 -06:00
commit 1ed34ad439
30 changed files with 5121 additions and 949 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,
}
}

63
client.go
View file

@ -45,8 +45,8 @@ func NewClient(host string) (*Client, error) {
// Host returns the host the client was initialized with.
func (c *Client) Host() string { return c.host }
// SliceN returns the number of slices on a server.
func (c *Client) SliceN(ctx context.Context) (uint64, error) {
// MaxSliceByDatabase returns the number of slices on a server by database.
func (c *Client) MaxSliceByDatabase(ctx context.Context) (map[string]uint64, error) {
// Execute request against the host.
u := url.URL{
Scheme: "http",
@ -57,24 +57,24 @@ func (c *Client) SliceN(ctx context.Context) (uint64, error) {
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return 0, err
return nil, err
}
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return 0, err
return nil, err
}
defer resp.Body.Close()
var rsp sliceMaxResponse
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("http: status=%d", resp.StatusCode)
return nil, fmt.Errorf("http: status=%d", resp.StatusCode)
} else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return 0, fmt.Errorf("json decode: %s", err)
return nil, fmt.Errorf("json decode: %s", err)
}
return rsp.SliceMax, nil
return rsp.MaxSlices, nil
}
// Schema returns all database and frame schema information.
@ -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)
}
@ -227,14 +227,16 @@ func MarshalImportPayload(db, frame string, slice uint64, bits []Bit) ([]byte, e
// Separate bitmap and profile IDs to reduce allocations.
bitmapIDs := Bits(bits).BitmapIDs()
profileIDs := Bits(bits).ProfileIDs()
timestamps := Bits(bits).Timestamps()
// 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,
Timestamps: timestamps,
})
if err != nil {
return nil, fmt.Errorf("marshal import request: %s", err)
@ -272,7 +274,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)
}
@ -362,13 +364,13 @@ func (c *Client) BackupTo(ctx context.Context, w io.Writer, db, frame string) er
tw := tar.NewWriter(w)
// Find the maximum number of slices.
sliceN, err := c.SliceN(ctx)
maxSlices, err := c.MaxSliceByDatabase(ctx)
if err != nil {
return fmt.Errorf("slice n: %s", err)
}
// Backup every slice to the tar file.
for i := uint64(0); i <= sliceN; i++ {
for i := uint64(0); i <= maxSlices[db]; i++ {
if err := c.backupSliceTo(ctx, tw, db, frame, i); err != nil {
return err
}
@ -644,10 +646,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
@ -779,6 +781,7 @@ func (c *Client) BitmapAttrDiff(ctx context.Context, db, frame string, blks []At
type Bit struct {
BitmapID uint64
ProfileID uint64
Timestamp int64
}
// Bits represents a slice of bits.
@ -789,6 +792,9 @@ func (p Bits) Len() int { return len(p) }
func (p Bits) Less(i, j int) bool {
if p[i].BitmapID == p[j].BitmapID {
if p[i].ProfileID < p[j].ProfileID {
return p[i].Timestamp < p[j].Timestamp
}
return p[i].ProfileID < p[j].ProfileID
}
return p[i].BitmapID < p[j].BitmapID
@ -812,6 +818,15 @@ func (a Bits) ProfileIDs() []uint64 {
return other
}
// Timestamps returns a slice of all the timestamps.
func (a Bits) Timestamps() []int64 {
other := make([]int64, len(a))
for i := range a {
other[i] = a[i].Timestamp
}
return other
}
// GroupBySlice returns a map of bits by slice.
func (a Bits) GroupBySlice() map[uint64][]Bit {
m := make(map[uint64][]Bit)
@ -834,5 +849,9 @@ type BitsByPos []Bit
func (p BitsByPos) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p BitsByPos) Len() int { return len(p) }
func (p BitsByPos) Less(i, j int) bool {
return Pos(p[i].BitmapID, p[i].ProfileID) < Pos(p[j].BitmapID, p[j].ProfileID)
p0, p1 := Pos(p[i].BitmapID, p[i].ProfileID), Pos(p[j].BitmapID, p[j].ProfileID)
if p0 == p1 {
return p[i].Timestamp < p[j].Timestamp
}
return p0 < p1
}

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()

2
cluster.go
View file

@ -31,7 +31,7 @@ func (a Nodes) Contains(n *Node) bool {
return false
}
// ContainsHost returns true if host matches on of the node's host.
// ContainsHost returns true if host matches one of the node's host.
func (a Nodes) ContainsHost(host string) bool {
for _, n := range a {
if n.Host == host {

2
cluster_test.go
View file

@ -41,7 +41,7 @@ func TestCluster_Partition(t *testing.T) {
c.PartitionN = partitionN
partitionID := c.Partition(db, slice)
if partitionID < 0 || partitionID > partitionN {
if partitionID < 0 || partitionID >= partitionN {
t.Errorf("partition out of range: slice=%d, p=%d, n=%d", slice, partitionID, partitionN)
}

39
cmd/pilosactl/main.go
View file

@ -315,13 +315,13 @@ func (cmd *ExportCommand) Run(ctx context.Context) error {
}
// Determine slice count.
sliceN, err := client.SliceN(ctx)
maxSlices, err := client.MaxSliceByDatabase(ctx)
if err != nil {
return err
}
// Export each slice.
for slice := uint64(0); slice <= sliceN; slice++ {
for slice := uint64(0); slice <= maxSlices[cmd.Database]; slice++ {
logger.Printf("exporting slice: %d", slice)
if err := client.ExportCSV(ctx, cmd.Database, cmd.Frame, slice, w); err != nil {
return err
@ -403,9 +403,10 @@ func (cmd *SortCommand) Run(ctx context.Context) error {
// Read rows as bits.
r := csv.NewReader(f)
r.FieldsPerRecord = -1
a := make([]pilosa.Bit, 0, 1000000)
for {
bitmapID, profileID, err := readCSVRow(r)
bitmapID, profileID, timestamp, err := readCSVRow(r)
if err == io.EOF {
break
} else if err == errBlank {
@ -413,7 +414,7 @@ func (cmd *SortCommand) Run(ctx context.Context) error {
} else if err != nil {
return err
}
a = append(a, pilosa.Bit{BitmapID: bitmapID, ProfileID: profileID})
a = append(a, pilosa.Bit{BitmapID: bitmapID, ProfileID: profileID, Timestamp: timestamp})
}
// Sort bits by position.
@ -426,8 +427,15 @@ func (cmd *SortCommand) Run(ctx context.Context) error {
// Write CSV to buffer.
buf = buf[:0]
buf = strconv.AppendUint(buf, bit.BitmapID, 10)
buf = append(buf, ',')
buf = strconv.AppendUint(buf, bit.ProfileID, 10)
if bit.Timestamp != 0 {
buf = append(buf, ',')
buf = append(buf, time.Unix(0, bit.Timestamp).UTC().Format(pilosa.TimeFormat)...)
}
buf = append(buf, '\n')
// Write to output.
@ -1608,33 +1616,42 @@ func serial(bs ...bench.Benchmark) bench.Benchmark {
}
// readCSVRow reads a bitmap/profile pair from a CSV row.
func readCSVRow(r *csv.Reader) (bitmapID, profileID uint64, err error) {
func readCSVRow(r *csv.Reader) (bitmapID, profileID uint64, timestamp int64, err error) {
// Read CSV row.
record, err := r.Read()
if err != nil {
return 0, 0, err
return 0, 0, 0, err
}
// Ignore blank rows.
if record[0] == "" {
return 0, 0, errBlank
return 0, 0, 0, errBlank
} else if len(record) < 2 {
return 0, 0, fmt.Errorf("bad column count: %d", len(record))
return 0, 0, 0, fmt.Errorf("bad column count: %d", len(record))
}
// Parse bitmap id.
bitmapID, err = strconv.ParseUint(record[0], 10, 64)
if err != nil {
return 0, 0, fmt.Errorf("invalid bitmap id: %q", record[0])
return 0, 0, 0, fmt.Errorf("invalid bitmap id: %q", record[0])
}
// Parse bitmap id.
profileID, err = strconv.ParseUint(record[1], 10, 64)
if err != nil {
return 0, 0, fmt.Errorf("invalid profile id: %q", record[1])
return 0, 0, 0, fmt.Errorf("invalid profile id: %q", record[1])
}
return bitmapID, profileID, nil
// Parse timestamp, if available.
if len(record) > 2 && record[2] != "" {
t, err := time.Parse(pilosa.TimeFormat, record[2])
if err != nil {
return 0, 0, 0, fmt.Errorf("invalid timestamp: %q", record[2])
}
timestamp = t.UnixNano()
}
return bitmapID, profileID, timestamp, nil
}
// errBlank indicates a blank row in a CSV file.

266
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,9 +20,16 @@ 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
// Max Slice on any node in the cluster, according to this node
remoteMaxSlice uint64
// Profile attribute storage and cache
profileAttrStore *AttrStore
@ -27,9 +39,10 @@ type DB struct {
// NewDB returns a new instance of DB.
func NewDB(path, name string) *DB {
return &DB{
path: path,
name: name,
frames: make(map[string]*Frame),
path: path,
name: name,
frames: make(map[string]*Frame),
remoteMaxSlice: 0,
profileAttrStore: NewAttrStore(filepath.Join(path, "data")),
@ -53,6 +66,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 +111,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()
@ -112,20 +169,51 @@ func (db *DB) Close() error {
return nil
}
// SliceN returns the max slice in the database.
func (db *DB) SliceN() uint64 {
// MaxSlice returns the max slice in the database according to this node.
func (db *DB) MaxSlice() uint64 {
if db == nil {
return 0
}
db.mu.Lock()
defer db.mu.Unlock()
var max uint64
max := db.remoteMaxSlice
for _, f := range db.frames {
if slice := f.SliceN(); slice > max {
if slice := f.MaxSlice(); slice > max {
max = slice
}
}
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 +275,144 @@ 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
}
// CreateFragmentIfNotExists returns a fragment in the database by name/slice.
func (db *DB) CreateFragmentIfNotExists(name string, slice uint64) (*Fragment, error) {
f, err := db.CreateFrameIfNotExists(name)
if err != nil {
return nil, err
}
return f.CreateFragmentIfNotExists(slice)
}
// 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
}
// Import bulk imports data.
func (db *DB) Import(name string, bitmapIDs, profileIDs []uint64, timestamps []*time.Time) error {
// Read frame.
f, err := db.CreateFrameIfNotExists(name)
if err != nil {
return err
}
// Determine quantum if timestamps are set.
var q TimeQuantum
if hasTime(timestamps) {
if q = f.TimeQuantum(); q == "" {
q = db.TimeQuantum()
if err := f.SetTimeQuantum(q); err != nil {
return err
}
}
if q == "" {
return errors.New("time quantum not set in either database or frame")
}
}
// Split import data by fragment.
dataByFragment := make(map[importKey]importData)
for i := range bitmapIDs {
bitmapID, profileID, timestamp := bitmapIDs[i], profileIDs[i], timestamps[i]
slice := profileID / SliceWidth
var names []string
if timestamp == nil {
names = []string{name}
} else {
names = FramesByTime(name, *timestamp, q)
}
// Attach bit to each frame.
for _, name := range names {
key := importKey{Frame: name, Slice: slice}
data := dataByFragment[key]
data.BitmapIDs = append(data.BitmapIDs, bitmapID)
data.ProfileIDs = append(data.ProfileIDs, profileID)
dataByFragment[key] = data
}
}
// Import into each fragment.
for key, data := range dataByFragment {
f, err := db.CreateFragmentIfNotExists(key.Frame, key.Slice)
if err != nil {
return err
}
if err := f.Import(data.BitmapIDs, data.ProfileIDs); err != nil {
return err
}
}
return nil
}
type dbSlice []*DB
func (p dbSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
@ -234,3 +460,29 @@ func MergeSchemas(a, b []*DBInfo) []*DBInfo {
return dbs
}
func (db *DB) SetRemoteMaxSlice(newmax uint64) {
db.mu.Lock()
defer db.mu.Unlock()
db.remoteMaxSlice = newmax
}
// hasTime returns true if a contains a non-nil time.
func hasTime(a []*time.Time) bool {
for _, t := range a {
if t != nil {
return true
}
}
return false
}
type importKey struct {
Frame string
Slice uint64
}
type importData struct {
BitmapIDs []uint64
ProfileIDs []uint64
}

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
}

69
executor.go
View file

@ -53,11 +53,10 @@ func (e *Executor) Execute(ctx context.Context, db string, q *pql.Query, slices
// If slices aren't specified, then include all of them.
if len(slices) == 0 {
// Round up the number of slices.
sliceN := e.Index.SliceN()
sliceN += (sliceN % uint64(len(e.Cluster.Nodes))) + uint64(len(e.Cluster.Nodes))
maxSlice := e.Index.DB(db).MaxSlice()
// Generate a slices of all slices.
slices = make([]uint64, sliceN+1)
slices = make([]uint64, maxSlice+1)
for i := range slices {
slices[i] = uint64(i)
}
@ -304,11 +303,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 +421,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 +594,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 +646,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 +662,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:
@ -714,7 +728,7 @@ func (e *Executor) mapReduce(ctx context.Context, db string, slices []uint64, c
// Iterate over all map responses and reduce.
var result interface{}
var sliceN int
var maxSlice int
for {
select {
case <-ctx.Done():
@ -739,8 +753,8 @@ func (e *Executor) mapReduce(ctx context.Context, db string, slices []uint64, c
result = reduceFn(result, resp.result)
// If all slices have been processed then return.
sliceN += len(resp.slices)
if sliceN >= len(slices) {
maxSlice += len(resp.slices)
if maxSlice >= len(slices) {
return result, nil
}
}
@ -798,7 +812,7 @@ func (e *Executor) mapperLocal(ctx context.Context, slices []uint64, mapFn mapFu
}
// Reduce results
var sliceN int
var maxSlice int
var result interface{}
for {
select {
@ -809,11 +823,11 @@ func (e *Executor) mapperLocal(ctx context.Context, slices []uint64, mapFn mapFu
return nil, resp.err
}
result = reduceFn(result, resp.result)
sliceN++
maxSlice++
}
// Exit once all slices are processed.
if sliceN == len(slices) {
if maxSlice == len(slices) {
return result, nil
}
}
@ -837,7 +851,6 @@ type mapResponse struct {
// ExecOptions represents an execution context for a single Execute() call.
type ExecOptions struct {
Timestamp *time.Time
Quantum TimeQuantum
Remote bool
}

76
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)
}
}
@ -281,7 +289,7 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) {
t.Fatalf("unexpected db: %s", db)
} else if query.String() != `Bitmap(id=10, frame=f)` {
t.Fatalf("unexpected query: %s", query.String())
} else if !reflect.DeepEqual(slices, []uint64{0, 2, 4}) {
} else if !reflect.DeepEqual(slices, []uint64{0}) { //TODO: this is incorrect because the calling node doesn't know about slice 2
t.Fatalf("unexpected slices: %+v", slices)
}
@ -390,7 +398,7 @@ func TestExecutor_Execute_Remote_TopN(t *testing.T) {
s.Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
if db != `d` {
t.Fatalf("unexpected db: %s", db)
} else if !reflect.DeepEqual(slices, []uint64{0, 2, 4, 6}) {
} else if !reflect.DeepEqual(slices, []uint64{0, 2}) {
t.Fatalf("unexpected slices: %+v", slices)
}

46
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 {
@ -870,7 +830,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
// Verify that there are an equal number of bitmap ids and profile ids.
if len(bitmapIDs) != len(profileIDs) {
return fmt.Errorf("mismatch of bitmap and profile len: %d != %d", len(bitmapIDs), len(profileIDs))
return fmt.Errorf("mismatch of bitmap/profile len: %d != %d", len(bitmapIDs), len(profileIDs))
}
// Disconnect op writer so we don't append updates.

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)
}
}

105
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
@ -58,8 +60,8 @@ func (f *Frame) Path() string { return f.path }
// BitmapAttrStore returns the attribute storage.
func (f *Frame) BitmapAttrStore() *AttrStore { return f.bitmapAttrStore }
// SliceN returns the max slice in the frame.
func (f *Frame) SliceN() uint64 {
// MaxSlice returns the max slice in the frame.
func (f *Frame) MaxSlice() uint64 {
f.mu.Lock()
defer f.mu.Unlock()
@ -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
}
@ -128,7 +134,46 @@ func (f *Frame) openFragments() error {
frag.BitmapAttrStore = f.bitmapAttrStore
f.fragments[frag.Slice()] = frag
f.stats.Count("sliceN", 1)
f.stats.Count("maxSlice", 1)
}
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
@ -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
// Persist 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))
@ -202,7 +275,7 @@ func (f *Frame) createFragmentIfNotExists(slice uint64) (*Fragment, error) {
// Save to lookup.
f.fragments[slice] = frag
f.stats.Count("sliceN", 1)
f.stats.Count("maxSlice", 1)
return frag, nil
}
@ -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 := profileID / 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
}

10
glide.lock generated
View file

@ -1,5 +1,5 @@
hash: 3b8279eaeec5a7b790ab8f2beacc451145362e91d8d9871d1c5b2313f7cd5d19
updated: 2016-12-20T13:41:45.723967488-06:00
hash: 469de49a1736f34a11e9b0e490f7c1da1d8cb0219fed4bf3ad9e71344ca7f58a
updated: 2017-01-09T16:52:45.035926627-06:00
imports:
- name: github.com/boltdb/bolt
version: 4b1ebc1869ad66568b313d0dc410e2be72670dda
@ -21,10 +21,14 @@ imports:
version: a6b377e3400b08991b80d6805d627f347f983866
subpackages:
- lru
- name: github.com/golang/protobuf
version: 8ee79997227bf9b34611aee7946ae64735e6fd93
subpackages:
- proto
- name: github.com/satori/go.uuid
version: 879c5887cd475cd7864858769793b2ceb0d44feb
- name: golang.org/x/crypto
version: d8e61c69ab46ca38328da2f4995abaf93b252290
version: c3b1d0d6d8690eaebe3064711b026770cc37efa3
subpackages:
- curve25519
- ed25519

3
glide.yaml
View file

@ -24,5 +24,6 @@ import:
version: c200b10b5d5e122be351b67af224adc6128af5bf
subpackages:
- unix
- package: github.com/golang/protobuf
- package: github.com/satori/go.uuid
version: v1.1.0
version: ^1.1.0

231
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":
@ -165,6 +186,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/nodes":
switch r.Method {
case "GET":
h.handleGetNodes(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/version":
h.handleVersion(w, r)
@ -203,7 +231,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,
}
@ -253,10 +280,10 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
}
func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) error {
sm := h.Index.SliceN()
ms := h.Index.MaxSlices()
if strings.Contains(r.Header.Get("Accept"), "application/x-protobuf") {
pb := &internal.SliceMaxResponse{
SliceMax: &sm,
pb := &internal.MaxSlicesResponse{
MaxSlices: ms,
}
if buf, err := proto.Marshal(pb); err != nil {
return err
@ -265,11 +292,13 @@ func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) erro
}
return nil
}
return json.NewEncoder(w).Encode(sliceMaxResponse{SliceMax: sm})
return json.NewEncoder(w).Encode(sliceMaxResponse{
MaxSlices: ms,
})
}
type sliceMaxResponse struct {
SliceMax uint64 `json:"SliceMax"`
MaxSlices map[string]uint64 `json:"MaxSlices"`
}
// handleDeleteDB handles DELETE /db request.
@ -299,6 +328,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 +426,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 +646,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,33 +712,41 @@ 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()
// Convert timestamps to time.Time.
timestamps := make([]*time.Time, len(req.Timestamps))
for i, ts := range req.Timestamps {
if ts == 0 {
continue
}
t := time.Unix(0, ts)
timestamps[i] = &t
}
// Validate that this handler owns the slice.
if !h.Cluster.OwnsFragment(h.Host, db, slice) {
mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, db, slice)
if !h.Cluster.OwnsFragment(h.Host, req.DB, req.Slice) {
mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, req.DB, req.Slice)
http.Error(w, mesg, http.StatusPreconditionFailed)
return
}
// Find the correct fragment.
h.logger().Println("importing:", db, frame, slice)
f, err := h.Index.CreateFragmentIfNotExists(db, frame, slice)
h.logger().Println("importing:", req.DB, req.Frame, req.Slice)
db, err := h.Index.CreateDBIfNotExists(req.DB)
if err != nil {
h.logger().Printf("fragment error: db=%s, frame=%s, slice=%d, err=%s", db, frame, slice, err)
h.logger().Printf("fragment error: db=%s, frame=%s, slice=%d, err=%s", req.DB, req.Frame, req.Slice, err)
http.Error(w, "fragment error", http.StatusInternalServerError)
return
}
h.logger().Println("Import into Fragment:", db, frame, slice, len(req.GetProfileIDs()))
// Import into fragment.
err = f.Import(req.GetBitmapIDs(), req.GetProfileIDs())
err = db.Import(req.Frame, req.BitmapIDs, req.ProfileIDs, timestamps)
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", req.DB, req.Frame, req.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 +892,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 +901,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.
@ -816,14 +975,15 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request)
}
// Determine the maximum number of slices.
sliceN, err := client.SliceN(r.Context())
maxSlices, err := client.MaxSliceByDatabase(r.Context())
if err != nil {
http.Error(w, "cannot determine remote slice count: "+err.Error(), http.StatusInternalServerError)
return
}
// Loop over each slice and import it if this node owns it.
for slice := uint64(0); slice <= sliceN; slice++ {
//travis
for slice := uint64(0); slice <= maxSlices[db]; slice++ {
// Ignore this slice if we don't own it.
if !h.Cluster.OwnsFragment(h.Host, db, slice) {
continue
@ -859,6 +1019,13 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request)
}
}
// handleGetNodes handles /nodes requests.
func (h *Handler) handleGetNodes(w http.ResponseWriter, r *http.Request) {
if err := json.NewEncoder(w).Encode(h.Cluster.Nodes); err != nil {
h.logger().Printf("write version response error: %s", err)
}
}
// handleGetVersion handles /version requests.
func (h *Handler) handleVersion(w http.ResponseWriter, r *http.Request) {
if err := json.NewEncoder(w).Encode(struct {
@ -919,16 +1086,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 +1145,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

125
handler_test.go
View file

@ -54,6 +54,30 @@ func TestHandler_Schema(t *testing.T) {
}
}
// Ensure the handler can return the maxslice map.
func TestHandler_MaxSlices(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d0", "f0", 1).MustSetBits(30, (1*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d0", "f0", 1).MustSetBits(30, (1*SliceWidth)+2)
idx.MustCreateFragmentIfNotExists("d0", "f0", 3).MustSetBits(30, (3*SliceWidth)+4)
idx.MustCreateFragmentIfNotExists("d1", "f1", 0).MustSetBits(40, (0*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d1", "f1", 0).MustSetBits(40, (0*SliceWidth)+2)
idx.MustCreateFragmentIfNotExists("d1", "f1", 0).MustSetBits(40, (0*SliceWidth)+8)
h := NewHandler()
h.Index = idx.Index
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max", nil))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"MaxSlices":{"d0":3,"d1":0}}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
// Ensure the handler can accept URL arguments.
func TestHandler_Query_Args_URL(t *testing.T) {
h := NewHandler()
@ -93,8 +117,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 +181,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 +256,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 +292,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 +313,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 +415,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 +477,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()

35
index.go
View file

@ -18,8 +18,7 @@ const DefaultCacheFlushInterval = 1 * time.Minute
// Index represents a container for fragments.
type Index struct {
mu sync.Mutex
remoteMax uint64
mu sync.Mutex
// Databases by name.
dbs map[string]*DB
@ -43,9 +42,8 @@ type Index struct {
// NewIndex returns a new instance of Index.
func NewIndex() *Index {
return &Index{
dbs: make(map[string]*DB),
remoteMax: 0,
closing: make(chan struct{}, 0),
dbs: make(map[string]*DB),
closing: make(chan struct{}, 0),
Stats: NopStatsClient,
@ -108,18 +106,13 @@ func (i *Index) Close() error {
return nil
}
// SliceN returns the highest slice across all frames.
func (i *Index) SliceN() uint64 {
i.mu.Lock()
defer i.mu.Unlock()
sliceN := i.remoteMax
for _, db := range i.dbs {
if n := db.SliceN(); n > sliceN {
sliceN = n
}
// MaxSlices returns MaxSlice map for all databases.
func (i *Index) MaxSlices() map[string]uint64 {
a := make(map[string]uint64)
for _, db := range i.DBs() {
a[db.Name()] = db.MaxSlice()
}
return sliceN
return a
}
// Schema returns schema data for all databases and frames.
@ -266,12 +259,6 @@ func (i *Index) CreateFragmentIfNotExists(db, frame string, slice uint64) (*Frag
return f.CreateFragmentIfNotExists(slice)
}
func (i *Index) SetMax(newmax uint64) {
i.mu.Lock()
defer i.mu.Unlock()
i.remoteMax = newmax
}
// monitorCacheFlush periodically flushes all fragment caches sequentially.
// This is run in a goroutine.
func (i *Index) monitorCacheFlush() {
@ -332,8 +319,6 @@ func (s *IndexSyncer) IsClosing() bool {
// SyncIndex compares the index on host with the local index and resolves differences.
func (s *IndexSyncer) SyncIndex() error {
sliceN := s.Index.SliceN()
// Iterate over schema in sorted order.
for _, di := range s.Index.Schema() {
// Verify syncer has not closed.
@ -357,7 +342,7 @@ func (s *IndexSyncer) SyncIndex() error {
return fmt.Errorf("frame sync error: db=%s, frame=%s, err=%s", di.Name, fi.Name, err)
}
for slice := uint64(0); slice <= sliceN; slice++ {
for slice := uint64(0); slice <= s.Index.DB(di.Name).MaxSlice(); slice++ {
// Ignore slices that this host doesn't own.
if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) {
continue

53
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,25 +93,26 @@ 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)
}
// Set highest slice.
idx0.SetMax(3)
idx0.DB("d").SetRemoteMaxSlice(1)
idx0.DB("y").SetRemoteMaxSlice(3)
// Set up syncer.
syncer := pilosa.IndexSyncer{
@ -119,6 +120,7 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
Host: cluster.Nodes[0].Host,
Cluster: cluster,
}
if err := syncer.SyncIndex(); err != nil {
t.Fatal(err)
}
@ -139,10 +141,13 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
}
f = idx.Fragment("d", "f0", 1)
a := f.Bitmap(9).Bits()
if !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a)
}
if a := f.Bitmap(9).Bits(); !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a)
}
f = idx.Fragment("y", "z", 3)
if a := f.Bitmap(10).Bits(); !reflect.DeepEqual(a, []uint64{(3 * SliceWidth) + 4, (3 * SliceWidth) + 5, (3 * SliceWidth) + 7}) {
t.Fatalf("unexpected bits(%d/y/z): %+v", i, a)
@ -185,6 +190,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

4086
internal/internal.pb.go
View file

File diff suppressed because it is too large Load diff

88
internal/internal.proto
View file

@ -1,30 +1,42 @@
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;
int64 Timestamp = 3;
}
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 +44,50 @@ 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;
repeated int64 Timestamps = 6;
}
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;
}
@ -82,6 +95,7 @@ message Cache {
repeated uint64 BitmapIDs = 1;
}
message SliceMaxResponse {
required uint64 SliceMax = 1;
message MaxSlicesResponse {
map<string, uint64> MaxSlices = 1;
}

16
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
}
@ -78,3 +75,6 @@ func decodeProfile(pb *internal.Profile) *Profile {
return p
}
// TimeFormat is the go-style time format used to parse string dates.
const TimeFormat = "2006-01-02T15:04"

22
pilosactl/import.go
View file

@ -12,6 +12,7 @@ import (
"os"
"strconv"
"strings"
"time"
"github.com/pilosa/pilosa"
)
@ -83,9 +84,10 @@ of the CSV file are grouped by slice for the most efficient import.
The format of the CSV file is:
BITMAPID,PROFILEID
BITMAPID,PROFILEID,[TIME]
The file should contain no headers.
The file should contain no headers. The TIME column is optional and can be
omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
`)
}
@ -135,6 +137,7 @@ func (cmd *ImportCommand) importPath(ctx context.Context, path string) error {
// Read rows as bits.
r := csv.NewReader(f)
r.FieldsPerRecord = -1
rnum := 0
for {
rnum++
@ -154,19 +157,32 @@ func (cmd *ImportCommand) importPath(ctx context.Context, path string) error {
return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record))
}
var bit pilosa.Bit
// Parse bitmap id.
bitmapID, err := strconv.ParseUint(record[0], 10, 64)
if err != nil {
return fmt.Errorf("invalid bitmap id on row %d: %q", rnum, record[0])
}
bit.BitmapID = bitmapID
// Parse bitmap id.
profileID, err := strconv.ParseUint(record[1], 10, 64)
if err != nil {
return fmt.Errorf("invalid profile id on row %d: %q", rnum, record[1])
}
bit.ProfileID = profileID
a = append(a, pilosa.Bit{BitmapID: bitmapID, ProfileID: profileID})
// Parse time, if exists.
if len(record) > 2 && record[2] != "" {
t, err := time.Parse(pilosa.TimeFormat, record[2])
if err != nil {
return fmt.Errorf("invalid timestamp on row %d: %q", rnum, record[2])
}
bit.Timestamp = t.UnixNano()
}
a = append(a, bit)
// If we've reached the buffer size then import bits.
if len(a) == cmd.BufferSize {

53
server.go
View file

@ -115,7 +115,7 @@ func (s *Server) Open() error {
// Start background monitoring.
s.wg.Add(2)
go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
go func() { defer s.wg.Done(); s.monitorMaxSlice() }()
go func() { defer s.wg.Done(); s.monitorMaxSlices() }()
return nil
}
@ -180,14 +180,14 @@ func (s *Server) monitorAntiEntropy() {
}
}
// monitorMaxSlice periodically pulls the highest slice from each node in the cluster.
func (s *Server) monitorMaxSlice() {
// monitorMaxSlices periodically pulls the highest slice from each node in the cluster.
func (s *Server) monitorMaxSlices() {
// Ignore if only one node in the cluster.
if len(s.Cluster.Nodes) <= 1 {
return
}
ticker := time.NewTicker(time.Second * time.Duration(s.PollingInterval))
ticker := time.NewTicker(s.PollingInterval)
defer ticker.Stop()
for {
@ -197,24 +197,34 @@ func (s *Server) monitorMaxSlice() {
case <-ticker.C:
}
oldmax := s.Index.SliceN()
newmax := oldmax
oldmaxslices := s.Index.MaxSlices()
for _, node := range s.Cluster.Nodes {
if s.Host != node.Host {
newslice, _ := checkMaxSlice(node.Host)
if newslice > newmax {
newmax = newslice
maxSlices, _ := checkMaxSlices(node.Host)
for db, newmax := range maxSlices {
// if we don't know about a db locally, create it
// so that the /schema endpoint can report it
if localdb := s.Index.DB(db); localdb != nil {
if newmax > oldmaxslices[db] {
oldmaxslices[db] = newmax
localdb.SetRemoteMaxSlice(newmax)
}
} else {
d, err := s.Index.CreateDBIfNotExists(db)
if err != nil {
s.logger().Printf("Failed to create DB locally: %s", db)
return
}
oldmaxslices[db] = newmax
d.SetRemoteMaxSlice(newmax)
}
}
}
}
if newmax > oldmax {
s.Index.SetMax(newmax)
}
}
}
func checkMaxSlice(hostport string) (uint64, error) {
func checkMaxSlices(hostport string) (map[string]uint64, error) {
// Create HTTP request.
req, err := http.NewRequest("GET", (&url.URL{
Scheme: "http",
@ -223,7 +233,7 @@ func checkMaxSlice(hostport string) (uint64, error) {
}).String(), nil)
if err != nil {
return 0, err
return nil, err
}
// Require protobuf encoding.
@ -233,28 +243,27 @@ func checkMaxSlice(hostport string) (uint64, error) {
// Send request to remote node.
resp, err := http.DefaultClient.Do(req)
if err != nil {
return 0, err
return nil, err
}
defer resp.Body.Close()
// Read response into buffer.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return 0, err
return nil, err
}
// Check status code.
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body)
return nil, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body)
}
// Decode response object.
pb := internal.SliceMaxResponse{}
pb := internal.MaxSlicesResponse{}
if err = proto.Unmarshal(body, &pb); err != nil {
return 0, err
return nil, err
}
return *pb.SliceMax, nil
return pb.MaxSlices, 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
}