featurebase/handler.go
Seebs 4d494f6699
shared/generic functionality for iterating holders
This is sort of large, but it's annoyingly difficult to
separate out.

The basic idea is to allow us to have a single holder-iterating
block of code, which is associated with the holder, that can be used
for various things, like the snapshot queue background scan, or
for inspect operations.

We invent the concept of a HolderFilter, which is a thing that
can decide what things in a holder it cares about, and a HolderOperator,
which can also process those things selectively.

In the process, we fix up a couple of subtle bugs in the
inspect logic; specifically, the assumption that the mapped flag could
tell you whether a container was modified by the ops log doesn't
work with mmap, so we have a shiny new flag which is used to track
that, internal to the roaring/container code.

All of this leads to the actual *point* of this exercise, which is
making it easier to create an /inspect endpoint which produces almost
the same data we'd have gotten from `pilosa inspect` on a data directory;
the distinction is that it doesn't try to identify the distinction
between data from disk and data from operations since the file was
loaded. Possibly it should, but it doesn't yet.

The snapshot queue is now implemented using the HolderOperator
design, which requires some subtle changes to how it works, but
overall makes it easier to follow the snapshot queue logic,
and also shares that logic with the way Inspect works.

The holder's snapshot queue is now provided by the server, in
a default environment.

The queueless snapshot queue no longer triggers snapshots on
enqueue -- it turns out that breaks badly, because a key
point about enqueueing a snapshot is that it's safe to do it
*during* a transaction on that fragment, and triggering a
snapshot during a transaction actually causes horrible errors
as the ops log ends up being the old file, which we close.
Related to this, we also need to prevent closed fragments from
trying to snapshot, so we track fragment openness when opening
or closing, and bail on trying to snapshot a fragment which is closed.

We also stop using the queueless snapshot queue during tests,
because that's a horrible idea.

We copy a little bit of the partition logic from the cluster code so
we don't have to expose it all, this lets us check whether the node
we're looking at is the one which should be primary for a given shard,
and if not, identify which node would be. This works only when
pointed at a data directory, for now.

The test cases for the holder have to be internal, because pilosa
doesn't export view/fragment, just Index/Field. This means that the
holder test cases can't just use the test/* package, so they duplicate
some of its logic, approximately.
2020-06-29 15:18:47 -04:00

317 lines
8.6 KiB
Go

// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
import (
"encoding/json"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
)
// QueryRequest represent a request to process a query.
type QueryRequest struct {
// Index to execute query against.
Index string
// The query string to parse and execute.
Query string
// The shards to include in the query execution.
// If empty, all shards are included.
Shards []uint64
// Return column attributes, if true.
ColumnAttrs bool
// Do not return row attributes, if true.
ExcludeRowAttrs bool
// Do not return columns, if true.
ExcludeColumns bool
// If true, indicates that query is part of a larger distributed query.
// If false, this request is on the originating node.
Remote bool
// Should we profile this query?
Profile bool
// Additional data associated with the query, in cases where there's
// row-style inputs for precomputed values.
EmbeddedData []*Row
}
// QueryResponse represent a response from a processed query.
type QueryResponse struct {
// Result for each top-level query call.
// The result type differs depending on the query; types
// include: Row, RowIdentifiers, GroupCounts, SignedRow,
// ValCount, Pair, Pairs, bool, uint64.
Results []interface{}
// Set of column attribute objects matching IDs returned in Result.
ColumnAttrSets []*ColumnAttrSet
// Error during parsing or execution.
Err error
// Profiling data, if any
Profile *tracing.Profile
}
// MarshalJSON marshals QueryResponse into a JSON-encoded byte slice
func (resp *QueryResponse) MarshalJSON() ([]byte, error) {
if resp.Err != nil {
return json.Marshal(struct {
Err string `json:"error"`
}{Err: resp.Err.Error()})
}
return json.Marshal(struct {
Results []interface{} `json:"results"`
ColumnAttrSets []*ColumnAttrSet `json:"columnAttrs,omitempty"`
Profile *tracing.Profile `json:"profile,omitempty"`
}{
Results: resp.Results,
ColumnAttrSets: resp.ColumnAttrSets,
Profile: resp.Profile,
})
}
// Handler is the interface for the data handler, a wrapper around
// Pilosa's data store.
type Handler interface {
Serve() error
Close() error
}
type nopHandler struct{}
func (n nopHandler) Serve() error {
return nil
}
func (n nopHandler) Close() error {
return nil
}
// NopHandler is a no-op implementation of the Handler interface.
var NopHandler Handler = nopHandler{}
// ImportValueRequest describes the import request structure
// for a value (BSI) import.
type ImportValueRequest struct {
Index string
IndexCreatedAt int64
Field string
FieldCreatedAt int64
// if Shard is MaxUint64 (an impossible shard value), this
// indicates that the column IDs may come from multiple shards.
Shard uint64
ColumnIDs []uint64
ColumnKeys []string
Values []int64
FloatValues []float64
StringValues []string
}
func (ivr *ImportValueRequest) Len() int { return len(ivr.ColumnIDs) }
func (ivr *ImportValueRequest) Less(i, j int) bool { return ivr.ColumnIDs[i] < ivr.ColumnIDs[j] }
func (ivr *ImportValueRequest) Swap(i, j int) {
ivr.ColumnIDs[i], ivr.ColumnIDs[j] = ivr.ColumnIDs[j], ivr.ColumnIDs[i]
if len(ivr.Values) > 0 {
ivr.Values[i], ivr.Values[j] = ivr.Values[j], ivr.Values[i]
} else if len(ivr.FloatValues) > 0 {
ivr.FloatValues[i], ivr.FloatValues[j] = ivr.FloatValues[j], ivr.FloatValues[i]
} else if len(ivr.StringValues) > 0 {
ivr.StringValues[i], ivr.StringValues[j] = ivr.StringValues[j], ivr.StringValues[i]
}
}
// Validate ensures that the payload of the request is valid.
func (ivr *ImportValueRequest) Validate() error {
return ivr.ValidateWithTimestamp(ivr.IndexCreatedAt, ivr.FieldCreatedAt)
}
// ValidateWithTimestamp ensures that the payload of the request is valid.
func (ivr *ImportValueRequest) ValidateWithTimestamp(indexCreatedAt, fieldCreatedAt int64) error {
if ivr.Index == "" || ivr.Field == "" {
return errors.Errorf("index and field required, but got '%s' and '%s'", ivr.Index, ivr.Field)
}
if len(ivr.ColumnIDs) != 0 && len(ivr.ColumnKeys) != 0 {
return errors.Errorf("must pass either column ids or keys, but not both")
}
var valueSetCount int
if len(ivr.Values) != 0 {
valueSetCount++
}
if len(ivr.FloatValues) != 0 {
valueSetCount++
}
if len(ivr.StringValues) != 0 {
valueSetCount++
}
if valueSetCount > 1 {
return errors.Errorf("must pass ints, floats, or strings but not multiple")
}
if ivr.IndexCreatedAt != 0 && ivr.FieldCreatedAt != 0 {
if ivr.IndexCreatedAt != indexCreatedAt || ivr.FieldCreatedAt != fieldCreatedAt {
return ErrPreconditionFailed
}
}
return nil
}
// ImportColumnAttrsRequest describes the import request structure
// for a ColumnAttr import
type ImportColumnAttrsRequest struct {
AttrKey string
ColumnIDs []uint64
AttrVals []string
Shard int64
Index string
IndexCreatedAt int64
}
// ImportRequest describes the import request structure
// for an import.
type ImportRequest struct {
Index string
IndexCreatedAt int64
Field string
FieldCreatedAt int64
Shard uint64
RowIDs []uint64
ColumnIDs []uint64
RowKeys []string
ColumnKeys []string
Timestamps []int64
}
// ValidateWithTimestamp ensures that the payload of the request is valid.
func (ir *ImportRequest) ValidateWithTimestamp(indexCreatedAt, fieldCreatedAt int64) error {
if ir.IndexCreatedAt != 0 && ir.FieldCreatedAt != 0 {
if ir.IndexCreatedAt != indexCreatedAt || ir.FieldCreatedAt != fieldCreatedAt {
return ErrPreconditionFailed
}
}
return nil
}
const (
RequestActionSet = "set"
RequestActionClear = "clear"
RequestActionOverwrite = "overwrite"
)
// ImportRoaringRequest describes the import request structure
// for an import containing roaring-encoded data.
type ImportRoaringRequest struct {
IndexCreatedAt int64
FieldCreatedAt int64
Clear bool
Action string // [set, clear, overwrite]
Block int
Views map[string][]byte
}
// ValidateWithTimestamp ensures that the payload of the request is valid.
func (irr *ImportRoaringRequest) ValidateWithTimestamp(indexCreatedAt, fieldCreatedAt int64) error {
if irr.IndexCreatedAt != 0 && irr.FieldCreatedAt != 0 {
if irr.IndexCreatedAt != indexCreatedAt || irr.FieldCreatedAt != fieldCreatedAt {
return ErrPreconditionFailed
}
}
return nil
}
// ImportResponse is the structured response of an import.
type ImportResponse struct {
Err string
}
// BlockDataRequest describes the structure of a request
// for fragment block data.
type BlockDataRequest struct {
Index string
Field string
View string
Shard uint64
Block uint64
}
// BlockDataResponse is the structured response of a block
// data request.
type BlockDataResponse struct {
RowIDs []uint64
ColumnIDs []uint64
}
// TranslateKeysRequest describes the structure of a request
// for a batch of key translations.
type TranslateKeysRequest struct {
Index string
Field string
Keys []string
}
// TranslateKeysResponse is the structured response of a key
// translation request.
type TranslateKeysResponse struct {
IDs []uint64
}
// TranslateIDsRequest describes the structure of a request
// for a batch of id translations.
type TranslateIDsRequest struct {
Index string
Field string
IDs []uint64
}
// TranslateIDsResponse is the structured response of a id
// translation request.
type TranslateIDsResponse struct {
Keys []string
}
// InspectRequestParams represents the parts of an InspectRequest that
// aren't generic holder filtering attributes.
type InspectRequestParams struct {
Containers bool // include container details
Checksum bool // perform checksums
}
// InspectRequest represents a request for a possibly-partial
// holder inspection, using a provided holder filter and inspect-specific
// parameters.
type InspectRequest struct {
HolderFilterParams
InspectRequestParams
}
// InspectResponse contains the structured results for an InspectRequest.
// It may some day be expanded to include metadata about views or indexes.
type InspectResponse struct {
Fragments []struct {
Index string
Field string
View string
Shard int64
Path string
Info *FragmentInfo
}
}