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

Merge branch 'master' into allow-backslash-cr-in-pql-strings

Browse source
This commit is contained in:
Yuce Tekol 2018-11-06 14:12:36 +03:00
commit 56ee728724
No known key found for this signature in database
GPG key ID: CB59E46D2FB90573
6 changed files with 374 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

171
api.go
View file

@ -690,7 +690,8 @@ func (api *API) FieldAttrDiff(_ context.Context, indexName string, fieldName str
// ImportOptions holds the options for the API.Import method.
type ImportOptions struct {
Clear bool
Clear bool
IgnoreKeyCheck bool
}
// ImportOption is a functional option type for API.Import.
@ -703,8 +704,15 @@ func OptImportOptionsClear(c bool) ImportOption {
}
}
func OptImportOptionsIgnoreKeyCheck(b bool) ImportOption {
return func(o *ImportOptions) error {
o.IgnoreKeyCheck = b
return nil
}
}
// Import bulk imports data into a particular index,field,shard.
func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOption) error {
func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOption) error {
if err := api.validate(apiImport); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -715,34 +723,71 @@ func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOpti
return errors.Wrap(err, "setting up import options")
}
index := api.holder.Index(req.Index)
if index == nil {
return newNotFoundError(ErrIndexNotFound)
}
field, err := api.indexField(req.Index, req.Field, req.Shard)
index, field, err := api.indexField(req.Index, req.Field, req.Shard)
if err != nil {
return errors.Wrap(err, "getting field")
return errors.Wrap(err, "getting index and field")
}
// Translate row keys.
if field.keys() {
if len(req.RowIDs) != 0 {
return errors.New("row ids cannot be used because field uses string keys")
// Unless explicitly ignoring key validation (meaning keys have been
// translated to ids in a previous step at the coordinator node), then
// check to see if keys need translation.
if !options.IgnoreKeyCheck {
// Translate row keys.
if field.keys() {
if len(req.RowIDs) != 0 {
return errors.New("row ids cannot be used because field uses string keys")
}
if req.RowIDs, err = api.holder.translateFile.TranslateRowsToUint64(index.Name(), field.Name(), req.RowKeys); err != nil {
return errors.Wrap(err, "translating rows")
}
}
if req.RowIDs, err = api.holder.translateFile.TranslateRowsToUint64(index.Name(), field.Name(), req.RowKeys); err != nil {
return errors.Wrap(err, "translating rows")
// Translate column keys.
if index.Keys() {
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
}
}
// For translated data, map the columnIDs to shards. If
// this node does not own the shard, forward to the node that does.
if index.Keys() || field.keys() {
m := make(map[uint64][]Bit)
for i, colID := range req.ColumnIDs {
shard := colID / ShardWidth
if _, ok := m[shard]; !ok {
m[shard] = make([]Bit, 0)
}
m[shard] = append(m[shard], Bit{
RowID: req.RowIDs[i],
ColumnID: colID,
Timestamp: req.Timestamps[i],
})
}
// Signal to the receiving nodes to ignore checking for key translation.
opts = append(opts, OptImportOptionsIgnoreKeyCheck(true))
var eg errgroup.Group
for shard, bits := range m {
// TODO: if local node owns this shard we don't need to go through the client
shard := shard
bits := bits
eg.Go(func() error {
return api.server.defaultClient.Import(ctx, req.Index, req.Field, shard, bits, opts...)
})
}
return eg.Wait()
}
}
// Translate column keys.
if index.Keys() {
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
}
// Validate shard ownership.
if err := api.validateShardOwnership(req.Index, req.Shard); err != nil {
return errors.Wrap(err, "validating shard ownership")
}
// Convert timestamps to time.Time.
@ -772,7 +817,7 @@ func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOpti
}
// ImportValue bulk imports values into a particular field.
func (api *API) ImportValue(_ context.Context, req *ImportValueRequest, opts ...ImportOption) error {
func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts ...ImportOption) error {
if err := api.validate(apiImportValue); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -783,26 +828,60 @@ func (api *API) ImportValue(_ context.Context, req *ImportValueRequest, opts ...
return errors.Wrap(err, "setting up import options")
}
index := api.holder.Index(req.Index)
if index == nil {
return newNotFoundError(ErrIndexNotFound)
}
field, err := api.indexField(req.Index, req.Field, req.Shard)
index, field, err := api.indexField(req.Index, req.Field, req.Shard)
if err != nil {
return errors.Wrap(err, "getting field")
return errors.Wrap(err, "getting index and field")
}
// Translate column keys.
if index.Keys() {
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
// Unless explicitly ignoring key validation (meaning keys have been
// translate to ids in a previous step at the coordinator node), then
// check to see if keys need translation.
if !options.IgnoreKeyCheck {
// Translate column keys.
if index.Keys() {
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
}
// For translated data, map the columnIDs to shards. If
// this node does not own the shard, forward to the node that does.
m := make(map[uint64][]FieldValue)
for i, colID := range req.ColumnIDs {
shard := colID / ShardWidth
if _, ok := m[shard]; !ok {
m[shard] = make([]FieldValue, 0)
}
m[shard] = append(m[shard], FieldValue{
Value: req.Values[i],
ColumnID: colID,
})
}
// Signal to the receiving nodes to ignore checking for key translation.
opts = append(opts, OptImportOptionsIgnoreKeyCheck(true))
var eg errgroup.Group
for shard, vals := range m {
// TODO: if local node owns this shard we don't need to go through the client
shard := shard
vals := vals
eg.Go(func() error {
return api.server.defaultClient.ImportValue(ctx, req.Index, req.Field, shard, vals, opts...)
})
}
return eg.Wait()
}
}
// Validate shard ownership.
if err := api.validateShardOwnership(req.Index, req.Shard); err != nil {
return errors.Wrap(err, "validating shard ownership")
}
// Import columnIDs into existence field.
if !options.Clear {
if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
@ -863,28 +942,32 @@ func (api *API) LongQueryTime() time.Duration {
return api.cluster.longQueryTime
}
func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Field, error) {
func (api *API) validateShardOwnership(indexName string, shard uint64) error {
// Validate that this handler owns the shard.
if !api.cluster.ownsShard(api.Node().ID, indexName, shard) {
api.server.logger.Printf("node %s does not own shard %d of index %s", api.Node().ID, shard, indexName)
return nil, ErrClusterDoesNotOwnShard
return ErrClusterDoesNotOwnShard
}
return nil
}
func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Index, *Field, error) {
api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, shard)
// Find the Index.
api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, shard)
index := api.holder.Index(indexName)
if index == nil {
api.server.logger.Printf("fragment error: index=%s, field=%s, shard=%d, err=%s", indexName, fieldName, shard, ErrIndexNotFound.Error())
return nil, newNotFoundError(ErrIndexNotFound)
return nil, nil, newNotFoundError(ErrIndexNotFound)
}
// Retrieve field.
field := index.Field(fieldName)
if field == nil {
api.server.logger.Printf("field error: index=%s, field=%s, shard=%d, err=%s", indexName, fieldName, shard, ErrFieldNotFound.Error())
return nil, ErrFieldNotFound
return nil, nil, ErrFieldNotFound
}
return field, nil
return index, field, nil
}
// SetCoordinator makes a new Node the cluster coordinator.

232
api_test.go Normal file
View file

@ -0,0 +1,232 @@
// 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_test
import (
"context"
"fmt"
"reflect"
"testing"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
func TestAPI_Import(t *testing.T) {
c := test.MustRunCluster(t, 2,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node1"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
)},
)
defer c.Close()
m0 := c[0]
m1 := c[1]
t.Run("RowIDColumnKey", func(t *testing.T) {
ctx := context.Background()
index := "rick"
field := "f"
_, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
if err != nil {
t.Fatalf("creating field: %v", err)
}
rowID := uint64(1)
timestamp := int64(0)
// Generate some keyed records.
rowIDs := []uint64{}
colKeys := []string{}
timestamps := []int64{}
for i := 1; i <= 10; i++ {
rowIDs = append(rowIDs, rowID)
timestamps = append(timestamps, timestamp)
colKeys = append(colKeys, fmt.Sprintf("col%d", i))
}
// Import data with keys to the coordinator (node0) and verify that it gets
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
req := &pilosa.ImportRequest{
Index: index,
Field: field,
Shard: 0,
RowIDs: rowIDs,
ColumnKeys: colKeys,
Timestamps: timestamps,
}
if err := m0.API.Import(ctx, req); err != nil {
t.Fatal(err)
}
pql := fmt.Sprintf("Row(%s=%d)", field, rowID)
// Query node0.
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil {
t.Fatal(err)
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
t.Fatalf("unexpected column keys: %+v", keys)
}
// Query node1.
if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil {
t.Fatal(err)
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
t.Fatalf("unexpected column keys: %+v", keys)
}
})
t.Run("RowKeyColumnID", func(t *testing.T) {
ctx := context.Background()
index := "rkci"
field := "f"
_, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100), pilosa.OptFieldKeys())
if err != nil {
t.Fatalf("creating field: %v", err)
}
rowKey := "rowkey"
// Generate some keyed records.
rowKeys := []string{rowKey, rowKey, rowKey}
colIDs := []uint64{1, 2, pilosa.ShardWidth + 1}
timestamps := []int64{0, 0, 0}
// Import data with keys to the coordinator (node0) and verify that it gets
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
req := &pilosa.ImportRequest{
Index: index,
Field: field,
Shard: 0,
RowKeys: rowKeys,
ColumnIDs: colIDs,
Timestamps: timestamps,
}
if err := m0.API.Import(ctx, req); err != nil {
t.Fatal(err)
}
pql := fmt.Sprintf("Row(%s=%s)", field, rowKey)
// Query node0.
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil {
t.Fatal(err)
} else if columns := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, colIDs) {
t.Fatalf("unexpected column ids: %+v", columns)
}
// Query node1.
if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil {
t.Fatal(err)
} else if columns := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, colIDs) {
t.Fatalf("unexpected column ids: %+v", columns)
}
})
}
func TestAPI_ImportValue(t *testing.T) {
c := test.MustRunCluster(t, 2,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node1"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
)},
)
defer c.Close()
m0 := c[0]
m1 := c[1]
t.Run("ValColumnKey", func(t *testing.T) {
ctx := context.Background()
index := "valck"
field := "f"
_, err := m0.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeInt(0, 100))
if err != nil {
t.Fatalf("creating field: %v", err)
}
// Generate some keyed records.
values := []int64{}
colKeys := []string{}
for i := 1; i <= 10; i++ {
values = append(values, int64(i))
colKeys = append(colKeys, fmt.Sprintf("col%d", i))
}
// Import data with keys to the coordinator (node0) and verify that it gets
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
req := &pilosa.ImportValueRequest{
Index: index,
Field: field,
ColumnKeys: colKeys,
Values: values,
}
if err := m0.API.ImportValue(ctx, req); err != nil {
t.Fatal(err)
}
pql := fmt.Sprintf("Range(%s>0)", field)
// Query node0.
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil {
t.Fatal(err)
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
t.Fatalf("unexpected column keys: %+v", keys)
}
// Query node1.
if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: pql}); err != nil {
t.Fatal(err)
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
t.Fatalf("unexpected column keys: %+v", keys)
}
})
}
// offsetModHasher represents a simple, mod-based hashing offset by 1.
type offsetModHasher struct{}
func (*offsetModHasher) Hash(key uint64, n int) int {
return int(key+1) % n
}

2
enterprise/b/btree.go
View file

@ -75,7 +75,7 @@ type (
// 0 if a == b
// > 0 if a > b
//
Cmp func(a, b uint64) int
Cmp func(a, b uint64) int64
d struct { // data page
c int

4
enterprise/b/containers_btree.go
View file

@ -23,8 +23,8 @@ import (
"github.com/pilosa/pilosa/roaring"
)
func cmp(a, b uint64) int {
return int(a - b)
func cmp(a, b uint64) int64 {
return int64(a - b)
}
type bTreeContainers struct {

3
http/client.go
View file

@ -437,6 +437,9 @@ func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, inde
if opts.Clear {
vals.Set("clear", "true")
}
if opts.IgnoreKeyCheck {
vals.Set("ignoreKeyCheck", "true")
}
url := fmt.Sprintf("%s?%s", u.String(), vals.Encode())
req, err := http.NewRequest("POST", url, bytes.NewReader(buf))

22
http/handler.go
View file

@ -181,7 +181,7 @@ func (h *Handler) populateValidators() {
h.validators["DeleteIndex"] = queryValidationSpecRequired()
h.validators["PostField"] = queryValidationSpecRequired()
h.validators["DeleteField"] = queryValidationSpecRequired()
h.validators["PostImport"] = queryValidationSpecRequired().Optional("clear")
h.validators["PostImport"] = queryValidationSpecRequired().Optional("clear", "ignoreKeyCheck")
h.validators["PostImportRoaring"] = queryValidationSpecRequired().Optional("remote", "clear")
h.validators["PostQuery"] = queryValidationSpecRequired().Optional("shards", "columnAttrs", "excludeRowAttrs", "excludeColumns")
h.validators["GetInfo"] = queryValidationSpecRequired()
@ -260,20 +260,10 @@ func newRouter(handler *Handler) *mux.Router {
router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client
router.HandleFunc("/internal/translate/data", handler.handleGetTranslateData).Methods("GET").Name("GetTranslateData")
// TODO: Apply MethodNotAllowed statuses to all endpoints.
// Ideally this would be automatic, as described in this (wontfix) ticket:
// https://github.com/gorilla/mux/issues/6
// For now we just do it for the most commonly used handler, /query
router.HandleFunc("/index/{index}/query", handler.methodNotAllowedHandler).Methods("GET")
router.Use(handler.queryArgValidator)
return router
}
func (h *Handler) methodNotAllowedHandler(w http.ResponseWriter, _ *http.Request) {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
// ServeHTTP handles an HTTP request.
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
defer func() {
@ -997,6 +987,12 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
// If the clear flag is true, treat the import as clear bits.
q := r.URL.Query()
doClear := q.Get("clear") == "true"
doIgnoreKeyCheck := q.Get("ignoreKeyCheck") == "true"
opts := []pilosa.ImportOption{
pilosa.OptImportOptionsClear(doClear),
pilosa.OptImportOptionsIgnoreKeyCheck(doIgnoreKeyCheck),
}
// Get index and field type to determine how to handle the
// import data.
@ -1030,7 +1026,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
return
}
if err := h.api.ImportValue(r.Context(), req, pilosa.OptImportOptionsClear(doClear)); err != nil {
if err := h.api.ImportValue(r.Context(), req, opts...); err != nil {
switch errors.Cause(err) {
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)
@ -1048,7 +1044,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
return
}
if err := h.api.Import(r.Context(), req, pilosa.OptImportOptionsClear(doClear)); err != nil {
if err := h.api.Import(r.Context(), req, opts...); err != nil {
switch errors.Cause(err) {
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)