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

Merge branch 'master' into wrap-errors-frame

Browse source
This commit is contained in:
alanbernstein 2018-05-10 17:54:05 -05:00 • committed by GitHub
commit 515b1bd418
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 132 additions and 133 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

134
api.go
View file

@ -90,14 +90,14 @@ func (api *API) validate(f apiMethod) error {
// Query parses a PQL query out of the request and executes it.
func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, error) {
if err := api.validate(apiQuery); err != nil {
return QueryResponse{}, errors.Wrap(err, "validate api method")
return QueryResponse{}, errors.Wrap(err, "validating api method")
}
resp := QueryResponse{}
q, err := pql.NewParser(strings.NewReader(req.Query)).Parse()
if err != nil {
return resp, err
return resp, errors.Wrap(err, "parsing")
}
execOpts := &ExecOptions{
Remote: req.Remote,
@ -106,7 +106,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
}
results, err := api.Executor.Execute(ctx, req.Index, q, req.Slices, execOpts)
if err != nil {
return resp, err
return resp, errors.Wrap(err, "executing")
}
resp.Results = results
@ -125,7 +125,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
// Retrieve column attributes across all calls.
columnAttrSets, err := api.readColumnAttrSets(api.Holder.Index(req.Index), columnIDs)
if err != nil {
return resp, err
return resp, errors.Wrap(err, "reading column attrs")
}
resp.ColumnAttrSets = columnAttrSets
}
@ -143,7 +143,7 @@ func (api *API) readColumnAttrSets(index *Index, ids []uint64) ([]*ColumnAttrSet
// Read attributes for column. Skip column if empty.
attrs, err := index.ColumnAttrStore().Attrs(id)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting attrs")
} else if len(attrs) == 0 {
continue
}
@ -158,13 +158,13 @@ func (api *API) readColumnAttrSets(index *Index, ids []uint64) ([]*ColumnAttrSet
// CreateIndex makes a new Pilosa index.
func (api *API) CreateIndex(ctx context.Context, indexName string, options IndexOptions) (*Index, error) {
if err := api.validate(apiCreateIndex); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Create index.
index, err := api.Holder.CreateIndex(indexName, options)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating index")
}
// Send the create index message to all nodes.
err = api.Broadcaster.SendSync(
@ -174,7 +174,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
})
if err != nil {
api.Logger.Printf("problem sending CreateIndex message: %s", err)
return nil, err
return nil, errors.Wrap(err, "sending CreateIndex message")
}
api.Holder.Stats.Count("createIndex", 1, 1.0)
return index, nil
@ -183,7 +183,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
// Index retrieves the named index.
func (api *API) Index(ctx context.Context, indexName string) (*Index, error) {
if err := api.validate(apiIndex); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
index := api.Holder.Index(indexName)
@ -197,13 +197,13 @@ func (api *API) Index(ctx context.Context, indexName string) (*Index, error) {
// nothing and returns no error.
func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
if err := api.validate(apiDeleteIndex); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Delete index from the holder.
err := api.Holder.DeleteIndex(indexName)
if err != nil {
return err
return errors.Wrap(err, "deleting index")
}
// Send the delete index message to all nodes.
err = api.Broadcaster.SendSync(
@ -212,7 +212,7 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
})
if err != nil {
api.Logger.Printf("problem sending DeleteIndex message: %s", err)
return err
return errors.Wrap(err, "sending DeleteIndex message")
}
api.Holder.Stats.Count("deleteIndex", 1, 1.0)
return nil
@ -221,7 +221,7 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
// CreateFrame makes the named frame in the named index with the given options.
func (api *API) CreateFrame(ctx context.Context, indexName string, frameName string, options FrameOptions) (*Frame, error) {
if err := api.validate(apiCreateFrame); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Find index.
@ -233,7 +233,7 @@ func (api *API) CreateFrame(ctx context.Context, indexName string, frameName str
// Create frame.
frame, err := index.CreateFrame(frameName, options)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating frame")
}
// Send the create frame message to all nodes.
@ -245,7 +245,7 @@ func (api *API) CreateFrame(ctx context.Context, indexName string, frameName str
})
if err != nil {
api.Logger.Printf("problem sending CreateFrame message: %s", err)
return nil, err
return nil, errors.Wrap(err, "sending CreateFrame message")
}
api.Holder.Stats.CountWithCustomTags("createFrame", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
return frame, nil
@ -256,7 +256,7 @@ func (api *API) CreateFrame(ctx context.Context, indexName string, frameName str
// action is taken.
func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName string) error {
if err := api.validate(apiDeleteFrame); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Find index.
@ -267,7 +267,7 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str
// Delete frame from the index.
if err := index.DeleteFrame(frameName); err != nil {
return err
return errors.Wrap(err, "deleting frame")
}
// Send the delete frame message to all nodes.
@ -278,7 +278,7 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str
})
if err != nil {
api.Logger.Printf("problem sending DeleteFrame message: %s", err)
return err
return errors.Wrap(err, "sending DeleteFrame message")
}
api.Holder.Stats.CountWithCustomTags("deleteFrame", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
return nil
@ -288,7 +288,7 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str
// CSV of the form <row>,<col>
func (api *API) ExportCSV(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, w io.Writer) error {
if err := api.validate(apiExportCSV); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Validate that this handler owns the slice.
@ -313,7 +313,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, frameName strin
strconv.FormatUint(columnID, 10),
})
}); err != nil {
return err
return errors.Wrap(err, "writing CSV")
}
// Ensure data is flushed.
@ -325,7 +325,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, frameName strin
// SliceNodes returns the node and all replicas which should contain a slice's data.
func (api *API) SliceNodes(ctx context.Context, indexName string, slice uint64) ([]*Node, error) {
if err := api.validate(apiSliceNodes); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
return api.Cluster.SliceNodes(indexName, slice), nil
@ -336,7 +336,7 @@ func (api *API) SliceNodes(ctx context.Context, indexName string, slice uint64)
// the UnmarshalFragment API call.
func (api *API) MarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) (io.WriterTo, error) {
if err := api.validate(apiMarshalFragment); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve fragment from holder.
@ -352,7 +352,7 @@ func (api *API) MarshalFragment(ctx context.Context, indexName string, frameName
// fragment's data.
func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, reader io.ReadCloser) error {
if err := api.validate(apiUnmarshalFragment); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame.
@ -364,18 +364,18 @@ func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameNa
// Retrieve view.
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
// Retrieve fragment from frame.
frag, err := view.CreateFragmentIfNotExists(slice)
if err != nil {
return err
return errors.Wrap(err, "creating fragment")
}
// Read fragment in from request body.
if _, err := frag.ReadFrom(reader); err != nil {
return err
return errors.Wrap(err, "reading fragment")
}
return nil
}
@ -385,7 +385,7 @@ func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameNa
// ids from a "block" which is a subdivision of a fragment.
func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, error) {
if err := api.validate(apiFragmentBlockData); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
reqBytes, err := ioutil.ReadAll(body)
@ -418,7 +418,7 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
// FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment.
func (api *API) FragmentBlocks(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) ([]FragmentBlock, error) {
if err := api.validate(apiFragmentBlocks); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve fragment from holder.
@ -436,7 +436,7 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName string, frameName
// from replicas in the cluster and restores that data to it.
func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName string, host *URI) error {
if err := api.validate(apiRestoreFrame); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Create a client for the remote cluster.
@ -445,7 +445,7 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
// Determine the maximum number of slices.
maxSlices, err := client.MaxSliceByIndex(ctx)
if err != nil {
return err
return errors.Wrap(err, "getting max slice")
}
// Retrieve frame.
@ -457,7 +457,7 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
// Retrieve list of all views.
views, err := client.FrameViews(ctx, indexName, frameName)
if err != nil {
return err
return errors.Wrap(err, "getting views")
}
// Loop over each slice and import it if this node owns it.
@ -472,19 +472,19 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
// Create view.
v, err := f.CreateViewIfNotExists(view)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
// Otherwise retrieve the local fragment.
frag, err := v.CreateFragmentIfNotExists(slice)
if err != nil {
return err
return errors.Wrap(err, "creating fragment")
}
// Stream backup from remote node.
rd, err := client.BackupSlice(ctx, indexName, frameName, view, slice)
if err != nil {
return err
return errors.Wrap(err, "getting backup")
} else if rd == nil {
continue // slice doesn't exist
}
@ -493,7 +493,7 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
if err := func() error {
defer rd.Close()
if _, err := frag.ReadFrom(rd); err != nil {
return err
return errors.Wrap(err, "reading fragment")
}
return nil
}(); err != nil {
@ -514,7 +514,7 @@ func (api *API) Hosts(ctx context.Context) []*Node {
// CreateInputDefinition is deprecated and will be removed. Do not use it.
func (api *API) CreateInputDefinition(ctx context.Context, indexName string, inputDefName string, inputDef InputDefinitionInfo) error {
if err := api.validate(apiCreateInputDefinition); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
api.Logger.Printf(`CreateInputDefinition is deprecated and will be removed.
@ -552,7 +552,7 @@ Please open an issue if you need to continue using it.`)
// InputDefinition is deprecated and will be removed.
func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefName string) (*InputDefinition, error) {
if err := api.validate(apiInputDefinition); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
api.Logger.Printf(`InputDefinition is deprecated and will be removed.`)
@ -572,7 +572,7 @@ func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefN
// DeleteInputDefinition is deprecated and will be removed.
func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inputDefName string) error {
if err := api.validate(apiDeleteInputDefinition); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
api.Logger.Printf("DeleteInputDefinition is deprecated and will be removed.")
@ -601,7 +601,7 @@ func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inp
// WriteInput is deprecated and will be removed.
func (api *API) WriteInput(ctx context.Context, indexName string, inputDefName string, reqs []interface{}) error {
if err := api.validate(apiWriteInput); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
api.Logger.Printf("WriteInput is deprecated and will be removed.")
@ -629,7 +629,7 @@ func (api *API) WriteInput(ctx context.Context, indexName string, inputDefName s
// RecalculateCaches forces all TopN caches to be updated. Used mainly for integration tests.
func (api *API) RecalculateCaches(ctx context.Context) error {
if err := api.validate(apiRecalculateCaches); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
err := api.Broadcaster.SendSync(&internal.RecalculateCaches{})
@ -644,7 +644,7 @@ func (api *API) RecalculateCaches(ctx context.Context) error {
// the body and forwards it to the BroadcastHandler.
func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
if err := api.validate(apiClusterMessage); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Read entire body.
@ -680,7 +680,7 @@ func (api *API) Schema(ctx context.Context) []*IndexInfo {
// CreateField creates a new BSI field in the given index and frame.
func (api *API) CreateField(ctx context.Context, indexName string, frameName string, field *Field) error {
if err := api.validate(apiCreateField); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame by name.
@ -691,7 +691,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, frameName str
// Create new field.
if err := f.CreateField(field); err != nil {
return err
return errors.Wrap(err, "creating field")
}
// Send the create field message to all nodes.
@ -704,13 +704,13 @@ func (api *API) CreateField(ctx context.Context, indexName string, frameName str
if err != nil {
api.Logger.Printf("problem sending CreateField message: %s", err)
}
return err
return errors.Wrap(err, "sending CreateField message")
}
// DeleteField deletes the given field.
func (api *API) DeleteField(ctx context.Context, indexName string, frameName string, fieldName string) error {
if err := api.validate(apiDeleteField); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame by name.
@ -721,7 +721,7 @@ func (api *API) DeleteField(ctx context.Context, indexName string, frameName str
// Delete field.
if err := f.DeleteField(fieldName); err != nil {
return err
return errors.Wrap(err, "deleting field")
}
// Send the delete field message to all nodes.
@ -734,13 +734,13 @@ func (api *API) DeleteField(ctx context.Context, indexName string, frameName str
if err != nil {
api.Logger.Printf("problem sending DeleteField message: %s", err)
}
return err
return errors.Wrap(err, "sending DeleteField message")
}
// Fields returns the fields in the given frame.
func (api *API) Fields(ctx context.Context, indexName string, frameName string) ([]*Field, error) {
if err := api.validate(apiFields); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
index := api.Holder.index(indexName)
@ -759,7 +759,7 @@ func (api *API) Fields(ctx context.Context, indexName string, frameName string)
// Views returns the views in the given frame.
func (api *API) Views(ctx context.Context, indexName string, frameName string) ([]*View, error) {
if err := api.validate(apiViews); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve views.
@ -776,7 +776,7 @@ func (api *API) Views(ctx context.Context, indexName string, frameName string) (
// DeleteView removes the given view.
func (api *API) DeleteView(ctx context.Context, indexName string, frameName string, viewName string) error {
if err := api.validate(apiDeleteView); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame.
@ -789,7 +789,7 @@ func (api *API) DeleteView(ctx context.Context, indexName string, frameName stri
if err := f.DeleteView(viewName); err != nil {
// Ignore this error because views do not exist on all nodes due to slice distribution.
if err != ErrInvalidView {
return err
return errors.Wrap(err, "deleting view")
}
}
@ -804,13 +804,13 @@ func (api *API) DeleteView(ctx context.Context, indexName string, frameName stri
api.Logger.Printf("problem sending DeleteView message: %s", err)
}
return err
return errors.Wrap(err, "sending DeleteView message")
}
// IndexAttrDiff
func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
if err := api.validate(apiIndexAttrDiff); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve index from holder.
@ -822,7 +822,7 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
// Retrieve local blocks.
localBlocks, err := index.ColumnAttrStore().Blocks()
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting blocks")
}
// Read all attributes from all mismatched blocks.
@ -831,7 +831,7 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
// Retrieve block data.
m, err := index.ColumnAttrStore().BlockData(blockID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting block")
}
// Copy to index-wide struct.
@ -844,7 +844,7 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
if err := api.validate(apiFrameAttrDiff); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve index from holder.
@ -856,7 +856,7 @@ func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName s
// Retrieve local blocks.
localBlocks, err := f.RowAttrStore().Blocks()
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting blocks")
}
// Read all attributes from all mismatched blocks.
@ -865,7 +865,7 @@ func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName s
// Retrieve block data.
m, err := f.RowAttrStore().BlockData(blockID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting block")
}
// Copy to index-wide struct.
@ -879,12 +879,12 @@ func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName s
// Import bulk imports data into a particular index,frame,slice.
func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
if err := api.validate(apiImport); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
_, frame, err := api.indexFrame(req.Index, req.Frame, req.Slice)
if err != nil {
return err
return errors.Wrap(err, "getting frame")
}
// Convert timestamps to time.Time.
@ -902,18 +902,18 @@ func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
if err != nil {
api.Logger.Printf("import error: index=%s, frame=%s, slice=%d, bits=%d, err=%s", req.Index, req.Frame, req.Slice, len(req.ColumnIDs), err)
}
return err
return errors.Wrap(err, "importing")
}
// ImportValue bulk imports values into a particular field.
func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest) error {
if err := api.validate(apiImportValue); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
_, frame, err := api.indexFrame(req.Index, req.Frame, req.Slice)
if err != nil {
return err
return errors.Wrap(err, "getting frame")
}
// Import into fragment.
@ -921,7 +921,7 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest
if err != nil {
api.Logger.Printf("import error: index=%s, frame=%s, slice=%d, field=%s, bits=%d, err=%s", req.Index, req.Frame, req.Slice, req.Field, len(req.ColumnIDs), err)
}
return err
return errors.Wrap(err, "importing")
}
// MaxSlices returns the maximum slice number for each index in a map.
@ -1049,7 +1049,7 @@ func (api *API) inputJSONDataParser(req map[string]interface{}, index *Index, na
// SetCoordinator makes a new Node the cluster coordinator.
func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode *Node, err error) {
if err := api.validate(apiSetCoordinator); err != nil {
return nil, nil, errors.Wrap(err, "validate api method")
return nil, nil, errors.Wrap(err, "validating api method")
}
oldNode = api.Cluster.NodeByID(api.Cluster.Coordinator)
@ -1079,7 +1079,7 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode
// removing the given node.
func (api *API) RemoveNode(id string) (*Node, error) {
if err := api.validate(apiRemoveNode); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
removeNode := api.Cluster.nodeByID(id)
@ -1098,7 +1098,7 @@ func (api *API) RemoveNode(id string) (*Node, error) {
// ResizeAbort stops the current resize job.
func (api *API) ResizeAbort() error {
if err := api.validate(apiResizeAbort); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
err := api.Cluster.CompleteCurrentJob(ResizeJobStateAborted)

110
fragment.go
View file

@ -21,7 +21,6 @@ import (
"container/heap"
"context"
"encoding/binary"
"errors"
"fmt"
"hash"
"io"
@ -42,6 +41,7 @@ import (
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/roaring"
"github.com/pkg/errors"
)
const (
@ -160,12 +160,12 @@ func (f *Fragment) Open() error {
if err := func() error {
// Initialize storage in a function so we can close if anything goes wrong.
if err := f.openStorage(); err != nil {
return err
return errors.Wrap(err, "opening storage")
}
// Fill cache with rows persisted to disk.
if err := f.openCache(); err != nil {
return err
return errors.Wrap(err, "opening cache")
}
// Clear checksums.
@ -206,7 +206,7 @@ func (f *Fragment) openStorage() error {
// If the file is empty then initialize it with an empty bitmap.
fi, err := f.file.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting file before")
} else if fi.Size() == 0 {
bi := bufio.NewWriter(f.file)
if _, err := f.storage.WriteTo(bi); err != nil {
@ -215,7 +215,7 @@ func (f *Fragment) openStorage() error {
bi.Flush()
fi, err = f.file.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting file after")
}
}
@ -298,13 +298,13 @@ func (f *Fragment) close() error {
// Flush cache if closing gracefully.
if err := f.flushCache(); err != nil {
f.Logger.Printf("fragment: error flushing cache on close: err=%s, path=%s", err, f.path)
return err
return errors.Wrap(err, "flushing cache")
}
// Close underlying storage.
if err := f.closeStorage(); err != nil {
f.Logger.Printf("fragment: error closing storage: err=%s, path=%s", err, f.path)
return err
return errors.Wrap(err, "closing storage")
}
// Remove checksums.
@ -393,12 +393,12 @@ func (f *Fragment) setBit(rowID, columnID uint64) (changed bool, err error) {
// Determine the position of the bit in the storage.
pos, err := f.pos(rowID, columnID)
if err != nil {
return false, err
return false, errors.Wrap(err, "getting bit ops")
}
// Write to storage.
if changed, err = f.storage.Add(pos); err != nil {
return false, err
return false, errors.Wrap(err, "writing")
}
// Don't update the cache if nothing changed.
@ -411,7 +411,7 @@ func (f *Fragment) setBit(rowID, columnID uint64) (changed bool, err error) {
// Increment number of operations until snapshot is required.
if err := f.incrementOpN(); err != nil {
return false, err
return false, errors.Wrap(err, "incrementing")
}
// Get the row from row cache or fragment.storage.
@ -445,12 +445,12 @@ func (f *Fragment) clearBit(rowID, columnID uint64) (changed bool, err error) {
// Determine the position of the bit in the storage.
pos, err := f.pos(rowID, columnID)
if err != nil {
return false, err
return false, errors.Wrap(err, "getting bit pos")
}
// Write to storage.
if changed, err = f.storage.Remove(pos); err != nil {
return false, err
return false, errors.Wrap(err, "writing")
}
// Don't update the cache if nothing changed.
@ -463,7 +463,7 @@ func (f *Fragment) clearBit(rowID, columnID uint64) (changed bool, err error) {
// Increment number of operations until snapshot is required.
if err := f.incrementOpN(); err != nil {
return false, err
return false, errors.Wrap(err, "incrementing")
}
// Get the row from cache or fragment.storage.
@ -493,7 +493,7 @@ func (f *Fragment) FieldValue(columnID uint64, bitDepth uint) (value uint64, exi
// If existence bit is unset then ignore remaining bits.
if v, err := f.bit(uint64(bitDepth), columnID); err != nil {
return 0, false, err
return 0, false, errors.Wrap(err, "getting existence bit")
} else if !v {
return 0, false, nil
}
@ -501,7 +501,7 @@ func (f *Fragment) FieldValue(columnID uint64, bitDepth uint) (value uint64, exi
// Compute other bits into a value.
for i := uint(0); i < bitDepth; i++ {
if v, err := f.bit(uint64(i), columnID); err != nil {
return 0, false, err
return 0, false, errors.Wrapf(err, "getting value bit %d", i)
} else if v {
value |= (1 << i)
}
@ -533,7 +533,7 @@ func (f *Fragment) SetFieldValue(columnID uint64, bitDepth uint, value uint64) (
// Mark value as set.
if c, err := f.setBit(uint64(bitDepth), columnID); err != nil {
return changed, err
return changed, errors.Wrap(err, "marking not-null")
} else if c {
changed = true
}
@ -548,20 +548,20 @@ func (f *Fragment) importSetFieldValue(columnID uint64, bitDepth uint, value uin
if value&(1<<i) != 0 {
bit, err := f.pos(uint64(i), columnID)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "getting set pos")
}
if c, err := f.storage.Add(bit); err != nil {
return changed, err
return changed, errors.Wrap(err, "adding")
} else if c {
changed = true
}
} else {
bit, err := f.pos(uint64(i), columnID)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "getting clear pos")
}
if c, err := f.storage.Remove(bit); err != nil {
return changed, err
return changed, errors.Wrap(err, "removing")
} else if c {
changed = true
}
@ -571,10 +571,10 @@ func (f *Fragment) importSetFieldValue(columnID uint64, bitDepth uint, value uin
// Mark value as set.
p, err := f.pos(uint64(bitDepth), columnID)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "marking not-null")
}
if c, err := f.storage.Add(p); err != nil {
return changed, err
return changed, errors.Wrap(err, "adding to storage")
} else if c {
changed = true
}
@ -945,7 +945,7 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
if filters != nil {
attr, err := f.RowAttrStore.Attrs(rowID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting attrs")
} else if attr == nil {
continue
} else if attrValue := attr[opt.FilterField]; attrValue == nil {
@ -1308,14 +1308,14 @@ func (f *Fragment) MergeBlock(id int, data []PairSet) (sets, clears []PairSet, e
// Set local bits.
for i := range sets[0].ColumnIDs {
if _, err := f.setBit(sets[0].RowIDs[i], (f.Slice()*SliceWidth)+sets[0].ColumnIDs[i]); err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "setting")
}
}
// Clear local bits.
for i := range clears[0].ColumnIDs {
if _, err := f.clearBit(clears[0].RowIDs[i], (f.Slice()*SliceWidth)+clears[0].ColumnIDs[i]); err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "clearing")
}
}
@ -1346,13 +1346,13 @@ func (f *Fragment) Import(rowIDs, columnIDs []uint64) error {
// Determine the position of the bit in the storage.
pos, err := f.pos(rowID, columnID)
if err != nil {
return err
return errors.Wrap(err, "getting bit pos")
}
// Write to storage.
_, err = f.storage.Add(pos)
if err != nil {
return err
return errors.Wrap(err, "writing")
}
// Reduce the StatsD rate for high volume stats
f.stats.Count("ImportBit", 1, 0.0001)
@ -1386,7 +1386,7 @@ func (f *Fragment) Import(rowIDs, columnIDs []uint64) error {
// Write the storage to disk and reload.
if err := f.snapshot(); err != nil {
return err
return errors.Wrap(err, "snapshotting")
}
return nil
@ -1410,7 +1410,7 @@ func (f *Fragment) ImportValue(columnIDs, values []uint64, bitDepth uint) error
_, err := f.importSetFieldValue(columnID, bitDepth, value)
if err != nil {
return err
return errors.Wrap(err, "setting")
}
}
return nil
@ -1420,7 +1420,7 @@ func (f *Fragment) ImportValue(columnIDs, values []uint64, bitDepth uint) error
return err
}
if err := f.snapshot(); err != nil {
return err
return errors.Wrap(err, "snapshotting")
}
return nil
}
@ -1525,12 +1525,12 @@ func (f *Fragment) flushCache() error {
// Marshal cache data to bytes.
buf, err := proto.Marshal(&internal.Cache{IDs: ids})
if err != nil {
return err
return errors.Wrap(err, "marshalling")
}
// Write to disk.
if err := ioutil.WriteFile(f.CachePath(), buf, 0666); err != nil {
return err
return errors.Wrap(err, "writing")
}
return nil
@ -1540,7 +1540,7 @@ func (f *Fragment) flushCache() error {
func (f *Fragment) WriteTo(w io.Writer) (n int64, err error) {
// Force cache flush.
if err := f.FlushCache(); err != nil {
return 0, err
return 0, errors.Wrap(err, "flushing cache")
}
// Write out data and cache to a tar archive.
@ -1558,7 +1558,7 @@ func (f *Fragment) writeStorageToArchive(tw *tar.Writer) error {
// Open separate file descriptor to read from.
file, err := os.Open(f.path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer file.Close()
@ -1571,7 +1571,7 @@ func (f *Fragment) writeStorageToArchive(tw *tar.Writer) error {
fi, err := file.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting")
}
sz = fi.Size()
@ -1587,13 +1587,13 @@ func (f *Fragment) writeStorageToArchive(tw *tar.Writer) error {
Size: sz,
ModTime: time.Now(),
}); err != nil {
return err
return errors.Wrap(err, "writing header")
}
// Copy the file up to the last known size.
// This is done outside the lock because the storage format is append-only.
if _, err := io.CopyN(tw, file, sz); err != nil {
return err
return errors.Wrap(err, "copying")
}
return nil
}
@ -1607,7 +1607,7 @@ func (f *Fragment) writeCacheToArchive(tw *tar.Writer) error {
if os.IsNotExist(err) {
return nil
} else if err != nil {
return err
return errors.Wrap(err, "reading cache")
}
// Write archive header.
@ -1617,12 +1617,12 @@ func (f *Fragment) writeCacheToArchive(tw *tar.Writer) error {
Size: int64(len(buf)),
ModTime: time.Now(),
}); err != nil {
return err
return errors.Wrap(err, "writing header")
}
// Write data to archive.
if _, err := tw.Write(buf); err != nil {
return err
return errors.Wrap(err, "writing")
}
return nil
}
@ -1639,18 +1639,18 @@ func (f *Fragment) ReadFrom(r io.Reader) (n int64, err error) {
if err == io.EOF {
break
} else if err != nil {
return 0, err
return 0, errors.Wrap(err, "opening")
}
// Process file based on file name.
switch hdr.Name {
case "data":
if err := f.readStorageFromArchive(tr); err != nil {
return 0, err
return 0, errors.Wrap(err, "reading storage")
}
case "cache":
if err := f.readCacheFromArchive(tr); err != nil {
return 0, err
return 0, errors.Wrap(err, "reading cache")
}
default:
return 0, fmt.Errorf("invalid fragment archive file: %s", hdr.Name)
@ -1665,28 +1665,28 @@ func (f *Fragment) readStorageFromArchive(r io.Reader) error {
path := f.path + CopyExt
file, err := os.Create(path)
if err != nil {
return err
return errors.Wrap(err, "creating directory")
}
defer file.Close()
// Copy reader into temporary path.
if _, err = io.Copy(file, r); err != nil {
return err
return errors.Wrap(err, "copying")
}
// Close current storage.
if err := f.closeStorage(); err != nil {
return err
return errors.Wrap(err, "closing")
}
// Move snapshot to data file location.
if err := os.Rename(path, f.path); err != nil {
return err
return errors.Wrap(err, "renaming")
}
// Reopen storage.
if err := f.openStorage(); err != nil {
return err
return errors.Wrap(err, "opening")
}
return nil
@ -1696,14 +1696,14 @@ func (f *Fragment) readCacheFromArchive(r io.Reader) error {
// Slurp data from reader and write to disk.
buf, err := ioutil.ReadAll(r)
if err != nil {
return err
return errors.Wrap(err, "reading")
} else if err := ioutil.WriteFile(f.CachePath(), buf, 0666); err != nil {
return err
return errors.Wrap(err, "writing")
}
// Re-open cache.
if err := f.openCache(); err != nil {
return err
return errors.Wrap(err, "opening")
}
return nil
@ -1785,7 +1785,7 @@ func (s *FragmentSyncer) SyncFragment() error {
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
blocks, err := client.FragmentBlocks(context.Background(), s.Fragment.Index(), s.Fragment.Frame(), s.Fragment.View(), s.Fragment.Slice())
if err != nil && err != ErrFragmentNotFound {
return err
return errors.Wrap(err, "getting blocks")
}
blockSets = append(blockSets, blocks)
@ -1864,7 +1864,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
// Only sync the standard block.
rowIDs, columnIDs, err := client.BlockData(context.Background(), f.Index(), f.Frame(), ViewStandard, f.Slice(), id)
if err != nil {
return err
return errors.Wrap(err, "getting block")
}
pairSets = append(pairSets, PairSet{
@ -1881,7 +1881,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
// Merge blocks together.
sets, clears, err := f.MergeBlock(id, pairSets)
if err != nil {
return err
return errors.Wrap(err, "merging")
}
// Write updates to remote blocks.
@ -1926,7 +1926,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
}
_, err := clients[i].Query(context.Background(), f.Index(), queryRequest)
if err != nil {
return err
return errors.Wrap(err, "executing")
}
}
}

14
handler_test.go
View file

@ -639,7 +639,7 @@ func TestHandler_Query_Err_JSON(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader(`Bitmap(id=100)`)))
if w.Code != http.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"marker"}`+"\n" {
} else if body := w.Body.String(); body != `{"error":"executing: marker"}`+"\n" {
t.Fatalf("unexpected body: %q", body)
}
}
@ -667,7 +667,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.Err; s != `marker` {
} else if s := resp.Err; s != `executing: marker` {
t.Fatalf("unexpected error: %s", s)
}
}
@ -699,7 +699,7 @@ func TestHandler_Query_ErrParse(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/idx0/query?slices=0,1", strings.NewReader("bad_fn(")))
if w.Code != http.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"expected comma, right paren, or identifier, found \"\" occurred at line 1, char 8"}`+"\n" {
} else if body := w.Body.String(); body != `{"error":"parsing: expected comma, right paren, or identifier, found \"\" occurred at line 1, char 8"}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -909,7 +909,7 @@ func TestHandler_Frame_AddField(t *testing.T) {
)
if err != nil {
t.Fatal(err)
} else if body := MustReadAll(resp.Body); string(body) != `validating field: invalid field type`+"\n" {
} else if body := MustReadAll(resp.Body); string(body) != `creating field: validating field: invalid field range`+"\n" {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)
@ -931,7 +931,7 @@ func TestHandler_Frame_AddField(t *testing.T) {
)
if err != nil {
t.Fatal(err)
} else if body := MustReadAll(resp.Body); string(body) != `validating field: invalid field range`+"\n" {
} else if body := MustReadAll(resp.Body); string(body) != `creating field: validating field: invalid field range`+"\n" {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)
@ -955,7 +955,7 @@ func TestHandler_Frame_AddField(t *testing.T) {
)
if err != nil {
t.Fatal(err)
} else if body := MustReadAll(resp.Body); string(body) != `field already exists`+"\n" {
} else if body := MustReadAll(resp.Body); string(body) != `creating field: field already exists`+"\n" {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)
@ -1021,7 +1021,7 @@ func TestHandler_Frame_DeleteField(t *testing.T) {
t.Fatal(err)
} else if body, err := ioutil.ReadAll(resp.Body); err != nil {
t.Fatal(err)
} else if strings.TrimSpace(string(body)) != `field not found` {
} else if strings.TrimSpace(string(body)) != `deleting field: field not found` {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)

2
holder_test.go
View file

@ -243,7 +243,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
}
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open fragment: slice=0, err=unmarshal storage") {
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open fragment: slice=0, err=opening storage: unmarshal storage") {
t.Fatalf("unexpected error: %s", err)
}
})

5
server/server_test.go
View file

@ -32,7 +32,6 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
"github.com/pkg/errors"
)
// Ensure program can process queries and maintain consistency.
@ -53,10 +52,10 @@ func TestMain_Set_Quick(t *testing.T) {
// Execute SetBit() commands.
for _, cmd := range cmds {
if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && errors.Cause(err) != pilosa.ErrIndexExists {
if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && !strings.Contains(err.Error(), "index already exists") {
t.Fatal(err)
}
if err := client.CreateFrame(context.Background(), "i", cmd.Frame, pilosa.FrameOptions{}); err != nil && err != pilosa.ErrFrameExists {
if err := client.CreateFrame(context.Background(), "i", cmd.Frame, pilosa.FrameOptions{}); err != nil && !strings.Contains(err.Error(), "frame already exists") {
t.Fatal(err)
}
if _, err := m.Query("i", "", fmt.Sprintf(`SetBit(row=%d, frame=%q, col=%d)`, cmd.ID, cmd.Frame, cmd.ColumnID)); err != nil {