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

Merge branch 'FB-900_singleNodeDeployment' of github.com:molecula/featurebase into FB-900_singleNodeDeployment

Browse source
This commit is contained in:
Souhaila Noor 2021-11-08 15:22:23 -06:00
commit 2607e2dfb1
22 changed files with 843 additions and 584 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
.gitlab/.gitlab-ci.yml
View file

@ -160,6 +160,21 @@ build for darwin arm64:
paths:
- featurebase_darwin_arm64
package for linux amd64:
stage: build
image: registry.gitlab.com/molecula/featurebase/builder:0.0.3
variables:
GOOS: "linux"
GOARCH: "amd64"
script:
- echo 'deb [trusted=yes] https://repo.goreleaser.com/apt/ /' | tee /etc/apt/sources.list.d/goreleaser.list
- apt update && apt install nfpm
- make package
artifacts:
paths:
- "*.deb"
- "*.rpm"
# Build a FB Docker image with CI/CD and push to the GitLab registry.
build container fb:
image: docker:stable

8
Makefile
View file

@ -44,6 +44,9 @@ clean:
vendor: go.mod
$(GO) mod vendor
version:
@echo $(VERSION)
# Run test suite
test:
$(GO) test ./... -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) -v
@ -126,6 +129,11 @@ release-sans-ui: check-clean
$(MAKE) release-build GOOS=linux GOARCH=amd64
$(MAKE) release-build GOOS=linux GOARCH=arm64
package:
go build -o featurebase ./cmd/featurebase
nfpm package --packager deb --target featurebase_$(VERSION_ID).deb
nfpm package --packager rpm --target featurebase_$(VERSION_ID).rpm
# try (e.g.) internal/clustertests/docker-compose-replication2.yml
DOCKER_COMPOSE=internal/clustertests/docker-compose.yml

148
api.go
View file

@ -1454,13 +1454,6 @@ func OptImportOptionsPresorted(b bool) ImportOption {
}
}
func optImportOptionsFullySorted(b bool) ImportOption {
return func(o *ImportOptions) error {
o.fullySorted = b
return nil
}
}
var ErrAborted = fmt.Errorf("error: update was aborted")
func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRecord, opts ...ImportOption) error {
@ -1491,14 +1484,20 @@ func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRec
qcx.StartAtomicWriteTx(Txo{Write: writable, Index: idx, Shard: req.Shard})
tot := 0
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
// BSIs (Values)
for _, ivr := range req.Ivr {
tot++
if simPowerLoss && tot > lossAfter {
return ErrAborted
}
opts0 := append(opts, OptImportOptionsClear(ivr.Clear))
err := api.ImportValueWithTx(ctx, qcx, ivr, opts0...)
subOpts := *options
subOpts.Clear = ivr.Clear
err = api.ImportValueWithTx(ctx, qcx, ivr, &subOpts)
if err != nil {
return errors.Wrap(err, "ImportAtomicRecord ImportValueWithTx")
}
@ -1510,8 +1509,9 @@ func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRec
if simPowerLoss && tot > lossAfter {
return ErrAborted
}
opts0 := append(opts, OptImportOptionsClear(ir.Clear))
err := api.ImportWithTx(ctx, qcx, ir, opts0...)
subOpts := *options
subOpts.Clear = ir.Clear
err := api.ImportWithTx(ctx, qcx, ir, &subOpts)
if err != nil {
return errors.Wrap(err, "ImportAtomicRecord ImportWithTx")
}
@ -1539,7 +1539,12 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts .
if req.Clear {
opts = addClearToImportOptions(opts)
}
err = api.ImportWithTx(ctx, qcx, req, opts...)
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
err = api.ImportWithTx(ctx, qcx, req, options)
if err != nil {
return err
}
@ -1547,7 +1552,7 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts .
}
// ImportWithTx bulk imports data into a particular index,field,shard.
func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, opts ...ImportOption) error {
func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.Import")
defer span.Finish()
@ -1564,11 +1569,6 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
return errors.Wrap(err, "validating import value request")
}
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
span.LogKV(
"index", req.Index,
"field", req.Field)
@ -1586,6 +1586,8 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
if req.RowIDs, err = api.cluster.translateFieldKeys(ctx, field, req.RowKeys, true); err != nil {
return errors.Wrapf(err, "translating field keys")
}
} else if len(req.RowKeys) != 0 {
return errors.New("value keys cannot be used because field uses integer IDs")
}
// Translate column keys.
@ -1597,44 +1599,36 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys, true); 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 idx.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)
}
bit := Bit{
RowID: req.RowIDs[i],
ColumnID: colID,
}
if len(req.Timestamps) > 0 {
bit.Timestamp = req.Timestamps[i]
}
m[shard] = append(m[shard], bit)
}
// 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()
// mark this request as having an unknown shard, meaning it will
// be sorted and served out to multiple nodes.
req.Shard = ^uint64(0)
} else if len(req.ColumnKeys) != 0 {
return errors.New("record keys cannot be used because field uses integer IDs")
}
}
// if you specify a shard of ^0, we try to split this out. If we did any
// key translation, we set it to ^0 already above.
if req.Shard == ^uint64(0) {
reqs := req.SortToShards()
// Signal to the receiving nodes to ignore checking for key translation.
options.IgnoreKeyCheck = true
var eg errgroup.Group
for _, subReq := range reqs {
// TODO: if local node owns this shard we don't need to go through the client
subReq := subReq
eg.Go(func() error {
return api.server.defaultClient.Import(ctx, qcx, subReq, options)
})
}
return eg.Wait()
}
// otherwise, this has to be a shard that we have, and everything has
// to be for that shard.
// Validate shard ownership.
if err := api.validateShardOwnership(req.Index, req.Shard); err != nil {
return errors.Wrap(err, "validating shard ownership")
@ -1649,8 +1643,6 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
}
// Import columnIDs into existence field.
// Note: req.Shard may not be the only shard imported into here,
// so don't expect it to be invariant.
if !options.Clear {
if err := importExistenceColumns(qcx, idx, req.ColumnIDs, req.Shard); err != nil {
api.server.logger.Errorf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
@ -1662,7 +1654,7 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
}
// Import into fragment.
err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, req.Shard, opts...)
err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, req.Shard, options)
if err != nil {
api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
return errors.Wrap(err, "importing")
@ -1676,11 +1668,16 @@ func (api *API) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueReque
if req.Clear {
opts = addClearToImportOptions(opts)
}
return api.ImportValueWithTx(ctx, qcx, req, opts...)
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
return api.ImportValueWithTx(ctx, qcx, req, options)
}
// ImportValueWithTx bulk imports values into a particular field.
func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValueRequest, opts ...ImportOption) (err0 error) {
func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) (err0 error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.ImportValue")
defer span.Finish()
@ -1688,6 +1685,15 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
return errors.Wrap(err, "validating api method")
}
numCols := len(req.ColumnIDs) + len(req.ColumnKeys)
numVals := len(req.Values) + len(req.FloatValues) + len(req.TimestampValues) + len(req.StringValues)
if numCols != numVals {
return errors.New(fmt.Sprintf("number of columns (%v) and number of values (%v) do not match", numCols, numVals))
}
if numCols == 0 {
return nil
}
idx, field, err := api.indexField(req.Index, req.Field, req.Shard)
if err != nil {
return errors.Wrap(err, fmt.Sprintf("getting index '%v' and field '%v'; shard=%v", req.Index, req.Field, req.Shard))
@ -1697,12 +1703,6 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
return errors.Wrap(err, "validating import value request")
}
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
idx, field, err = api.indexField(req.Index, req.Field, req.Shard)
if err != nil {
return errors.Wrap(err, "getting index and field")
@ -1798,7 +1798,6 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
return errors.Wrap(err, "importing value")
} // end if req.Shard != math.MaxUint64
options.IgnoreKeyCheck = true
start := 0
shard := req.ColumnIDs[0] / ShardWidth
@ -1818,7 +1817,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
}
eg.Go(func() error {
return api.server.defaultClient.ImportValue2(ctx, subreq, options)
return api.server.defaultClient.ImportValue(ctx, qcx, subreq, options)
})
start = i
shard = colID / ShardWidth
@ -1839,7 +1838,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
// TODO we should elevate the logic for figuring out which
// node(s) to send to into API instead of having those details
// in the client implementation.
return api.server.defaultClient.ImportValue2(ctx, subreq, options)
return api.server.defaultClient.ImportValue(ctx, qcx, subreq, options)
})
err = eg.Wait()
if err != nil {
@ -2010,7 +2009,6 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha
// For each operation, we may have a set of records/fields to clear, and then
// also a set of fields to set/remove specific bits in.
opts := &ImportOptions{Presorted: true, IgnoreKeyCheck: true, fullySorted: true}
funcOpts := []ImportOption{OptImportOptionsIgnoreKeyCheck(true), OptImportOptionsPresorted(true), optImportOptionsFullySorted(true), OptImportOptionsClear(true)}
for _, op := range ops {
// ClearRecordIDs should exist only for delete, clear, and write. For clear and write,
// we'll have a list of fields, for delete, it should be all the fields.
@ -2059,12 +2057,6 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha
}
}
opts.Clear = (op.OpType == ingest.OpRemove)
// reslice this rather than regenerating it. i'm so efficient.
if opts.Clear {
funcOpts = funcOpts[:4]
} else {
funcOpts = funcOpts[:3]
}
// for "set" and "write" ops, we'll be setting bits, for
// "remove" ops we'll be clearing them, and for "clear" ops
// there shouldn't be anything here.
@ -2080,7 +2072,7 @@ func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, sha
}
switch field.Type() {
case "set", "time", "mutex", "bool":
err = field.Import(qcx, fieldOp.Values, fieldOp.RecordIDs, fieldOp.Signed, shard, funcOpts...)
err = field.Import(qcx, fieldOp.Values, fieldOp.RecordIDs, fieldOp.Signed, shard, opts)
case "int", "timestamp", "decimal":
err = field.importValue(qcx, fieldOp.RecordIDs, fieldOp.Signed, shard, opts)
default:
@ -2107,7 +2099,8 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard ui
// existence field write.
columnCopy := make([]uint64, len(columnIDs))
copy(columnCopy, columnIDs)
return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard)
options := ImportOptions{}
return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options)
}
func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error {
@ -2123,7 +2116,8 @@ func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uin
// existence field write.
columnCopy := make([]uint64, len(columnIDs))
copy(columnCopy, columnIDs)
return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, OptImportOptionsClear(true))
options := ImportOptions{Clear: true}
return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, &options)
}
// ShardDistribution returns an object representing the distribution of shards

217
api_test.go
View file

@ -22,6 +22,7 @@ import (
"math"
"math/rand"
"reflect"
"sort"
"strings"
"testing"
"time"
@ -64,55 +65,53 @@ func TestAPI_Import(t *testing.T) {
m0 := c.GetNode(0)
m1 := c.GetNode(1)
t.Run("RowIDColumnKey", func(t *testing.T) {
ctx := context.Background()
indexName := "rick"
fieldName := "f"
indexNames := map[bool]string{false: "i", true: "ki"}
fieldNames := map[bool]string{false: "f", true: "kf"}
index, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true})
ctx := context.Background()
for ik, indexName := range indexNames {
_, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: ik, TrackExistence: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
if index.CreatedAt() == 0 {
t.Fatal("index createdAt is empty")
for fk, fieldName := range fieldNames {
if fk {
_, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100), pilosa.OptFieldKeys())
} else {
_, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
}
if err != nil {
t.Fatalf("creating field: %v", err)
}
}
}
field, err := m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
if err != nil {
t.Fatalf("creating field: %v", err)
}
if field.CreatedAt() == 0 {
t.Fatal("field createdAt is empty")
}
N := 10
rowID := uint64(1)
timestamp := int64(0)
// Generate some keyed records.
rowIDs := []uint64{}
timestamps := []int64{}
N := 10
for i := 1; i <= N; i++ {
rowIDs = append(rowIDs, rowID)
timestamps = append(timestamps, timestamp)
}
// Keys are sharded so ordering is not guaranteed.
colKeys := []string{"col10", "col8", "col9", "col6", "col7", "col4", "col5", "col2", "col3", "col1"}
colKeys = colKeys[:N]
// Keys are sharded so ordering is not guaranteed.
colKeys := make([]string, N)
rowKeys := make([]string, N)
rowIDs := make([]uint64, N)
colIDs := make([]uint64, N)
for i := range colKeys {
colKeys[i] = fmt.Sprintf("col%d", i)
rowKeys[i] = fmt.Sprintf("row%d", i)
colIDs[i] = (uint64(i) + 1) * 3
rowIDs[i] = 1
}
sort.Strings(colKeys)
sort.Strings(rowKeys)
t.Run("RowIDColumnKey", func(t *testing.T) {
// Import data with keys to the primary and verify that it gets
// translated and forwarded to the owner of shard 0 (node1; because of offsetModHasher)
req := &pilosa.ImportRequest{
Index: indexName,
IndexCreatedAt: index.CreatedAt(),
Field: fieldName,
FieldCreatedAt: field.CreatedAt(),
Shard: 0, // import is all on shard 0, why are we making lots of other shards? b/c this is not a restriction.
RowIDs: rowIDs,
ColumnKeys: colKeys,
Timestamps: timestamps,
Index: indexNames[true],
Field: fieldNames[false],
Shard: 0, // inaccurate, but keys override it
RowIDs: rowIDs,
ColumnKeys: colKeys,
}
qcx := m0.API.Txf().NewQcx()
@ -122,20 +121,30 @@ func TestAPI_Import(t *testing.T) {
}
PanicOn(qcx.Finish())
pql := fmt.Sprintf("Row(%s=%d)", fieldName, rowID)
pql := fmt.Sprintf("Row(%s=%d)", fieldNames[false], rowIDs[0])
// Query node0.
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: indexName, Query: pql}); err != nil {
var keys []string
if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: indexNames[true], Query: pql}); err != nil {
t.Fatal(err)
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
} else {
keys = res.Results[0].(*pilosa.Row).Keys
}
sort.Strings(keys)
if !reflect.DeepEqual(keys, colKeys) {
t.Fatalf("expected colKeys='%#v'; observed column keys: %#v", colKeys, keys)
}
// Query node1.
if err := test.RetryUntil(5*time.Second, func() error {
if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: indexName, Query: pql}); err != nil {
if res, err := m1.API.Query(ctx, &pilosa.QueryRequest{Index: indexNames[true], Query: pql}); err != nil {
return err
} else if keys := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(keys, colKeys) {
} else {
keys = res.Results[0].(*pilosa.Row).Keys
}
sort.Strings(keys)
if !reflect.DeepEqual(keys, colKeys) {
return fmt.Errorf("unexpected column keys: %#v", keys)
}
return nil
@ -143,6 +152,57 @@ func TestAPI_Import(t *testing.T) {
t.Fatal(err)
}
})
t.Run("ExpectedErrors", func(t *testing.T) {
ctx := context.Background()
for ik, indexName := range indexNames {
for fk, fieldName := range fieldNames {
req := pilosa.ImportRequest{
Index: indexName,
Field: fieldName,
Shard: 0,
}
for rik := range indexNames {
if rik {
req.ColumnKeys = colKeys
req.ColumnIDs = nil
} else {
req.ColumnKeys = nil
req.ColumnIDs = colIDs
}
for rfk := range fieldNames {
if rfk {
req.RowKeys = rowKeys
req.RowIDs = nil
} else {
req.RowKeys = nil
req.RowIDs = rowIDs
}
err := func() error {
qcx := m0.API.Txf().NewQcx()
defer qcx.Abort()
err := m0.API.Import(ctx, qcx, req.Clone())
e2 := qcx.Finish()
if e2 != nil {
t.Fatalf("unexpected error committing: %v", e2)
}
return err
}()
if err != nil {
if rfk == fk && rik == ik {
t.Errorf("unexpected error: schema keys %t/%t, req keys %t/%t: %v",
ik, fk, rik, rfk, err)
}
} else {
if rfk != fk || rik != ik {
t.Errorf("unexpected no error: schema keys %t/%t, req keys %t/%t, req %#v",
ik, fk, rik, rfk, req)
}
}
}
}
}
}
})
// Relies on the previous test creating an index with TrackExistence and
// adding some data.
@ -221,6 +281,7 @@ func TestAPI_ImportValue(t *testing.T) {
Field: field,
ColumnKeys: colKeys,
Values: values,
Shard: 0, // inaccurate but keys override it
}
qcx := coord.API.Txf().NewQcx()
@ -251,6 +312,60 @@ func TestAPI_ImportValue(t *testing.T) {
}
})
t.Run("ValIntEmpty", func(t *testing.T) {
ctx := context.Background()
index := "valintempty"
field := "fld"
createIndexForTest(index, coord, t)
createFieldForTest(index, field, coord, t)
// Column keys are sharded so their order is not guaranteed.
colKeys := []string{"col2", "col1", "col3"}
values := []int64{1, 2, 3, 4}
// Import without data, verify that it succeeds
req := &pilosa.ImportValueRequest{
Index: index,
Field: field,
}
qcx1 := coord.API.Txf().NewQcx()
defer qcx1.Abort()
// Import with empty request, should succeed
if err := coord.API.ImportValue(ctx, qcx1, req); err != nil {
t.Fatal(err)
}
PanicOn(qcx1.Finish())
// Import without data but with columnkeys, verify that it errors
req.ColumnKeys = colKeys
qcx2 := coord.API.Txf().NewQcx()
defer qcx2.Abort()
if err := coord.API.ImportValue(ctx, qcx2, req); err == nil {
t.Fatal("expected error but succeeded")
}
PanicOn(qcx2.Finish())
// Import with mismatch column and value lengths
req.Values = values
qcx3 := coord.API.Txf().NewQcx()
defer qcx3.Abort()
if err := coord.API.ImportValue(ctx, qcx3, req); err == nil {
t.Fatal("expected error but succeeded")
}
PanicOn(qcx3.Finish())
// Import with data but no columns
req.ColumnKeys = make([]string, 0)
qcx4 := coord.API.Txf().NewQcx()
defer qcx4.Abort()
if err := coord.API.ImportValue(ctx, qcx4, req); err == nil {
t.Fatal("expected error but succeeded")
}
PanicOn(qcx4.Finish())
})
t.Run("ValDecimalField", func(t *testing.T) {
ctx := context.Background()
index := "valdec"
@ -1225,3 +1340,19 @@ func TestAPI_MutexCheck(t *testing.T) {
})
}
}
func createIndexForTest(index string, coord *test.Command, t *testing.T) {
ctx := context.Background()
_, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
}
func createFieldForTest(index string, field string, coord *test.Command, t *testing.T) {
ctx := context.Background()
_, err := coord.API.CreateField(ctx, index, field, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatalf("creating field: %v", err)
}
}

27
client.go
View file

@ -63,14 +63,11 @@ type InternalClient interface {
PartitionNodes(ctx context.Context, partitionID int) ([]*topology.Node, error)
Nodes(ctx context.Context) ([]*topology.Node, error)
Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error)
Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error
ImportK(ctx context.Context, index, field string, bits []Bit, opts ...ImportOption) error
Import(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error
EnsureIndex(ctx context.Context, name string, options IndexOptions) error
EnsureField(ctx context.Context, indexName string, fieldName string) error
EnsureFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error
ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue, opts ...ImportOption) error
ImportValueK(ctx context.Context, index, field string, vals []FieldValue, opts ...ImportOption) error
ImportValue2(ctx context.Context, req *ImportValueRequest, options *ImportOptions) error
ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error
ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error
CreateField(ctx context.Context, index, field string) error
CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error
@ -98,6 +95,10 @@ type InternalClient interface {
ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error
ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error
// SetInternalAPI tells the client the API it should use for internal/loopback ops
// where applicable.
SetInternalAPI(api *API)
}
//===============
@ -202,13 +203,10 @@ func (n nopInternalClient) Nodes(ctx context.Context) ([]*topology.Node, error)
func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n nopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error {
func (n nopInternalClient) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, options *ImportOptions) error {
return nil
}
func (n nopInternalClient) ImportK(ctx context.Context, index, field string, bits []Bit, opts ...ImportOption) error {
return nil
}
func (n nopInternalClient) ImportValue2(ctx context.Context, req *ImportValueRequest, options *ImportOptions) error {
func (n nopInternalClient) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueRequest, options *ImportOptions) error {
return nil
}
@ -249,12 +247,6 @@ func (n nopInternalClient) EnsureField(ctx context.Context, indexName string, fi
func (n nopInternalClient) EnsureFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error {
return nil
}
func (n nopInternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue, opts ...ImportOption) error {
return nil
}
func (n nopInternalClient) ImportValueK(ctx context.Context, index, field string, vals []FieldValue, opts ...ImportOption) error {
return nil
}
func (n nopInternalClient) ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error {
return nil
}
@ -305,3 +297,6 @@ func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, i
func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error {
return nil
}
func (c nopInternalClient) SetInternalAPI(api *API) {
}

95
ctl/import.go
View file

@ -19,15 +19,12 @@ import (
"encoding/csv"
"fmt"
"io"
"log"
"math"
"os"
"sort"
"strconv"
"time"
"github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/pql"
"github.com/molecula/featurebase/v2/server"
"github.com/pkg/errors"
@ -175,8 +172,6 @@ func (cmd *ImportCommand) importPath(ctx context.Context, fieldType string, useC
// bufferBits buffers slices of bits to be imported as a batch.
func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowKeys bool, path string) error {
a := make([]pilosa.Bit, 0, cmd.BufferSize)
var r *csv.Reader
if path != "-" {
@ -195,6 +190,13 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowK
r.FieldsPerRecord = -1
rnum := 0
req := &pilosa.ImportRequest{
Index: cmd.Index,
Field: cmd.Field,
Shard: ^uint64(0),
}
batchRecs := 0 // records in this batch
lastTime := 0 // last record that had a timestamp
for {
rnum++
@ -213,25 +215,28 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowK
return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record))
}
var bit pilosa.Bit
// Parse row id.
if useRowKeys {
bit.RowKey = record[0]
req.RowKeys = append(req.RowKeys, record[0])
} else {
if bit.RowID, err = strconv.ParseUint(record[0], 10, 64); err != nil {
var id uint64
if id, err = strconv.ParseUint(record[0], 10, 64); err != nil {
return fmt.Errorf("invalid row id on row %d: %q", rnum, record[0])
}
req.RowIDs = append(req.RowIDs, id)
}
// Parse column id.
if useColumnKeys {
bit.ColumnKey = record[1]
req.ColumnKeys = append(req.ColumnKeys, record[1])
} else {
if bit.ColumnID, err = strconv.ParseUint(record[1], 10, 64); err != nil {
var id uint64
if id, err = strconv.ParseUint(record[1], 10, 64); err != nil {
return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1])
}
req.ColumnIDs = append(req.ColumnIDs, id)
}
batchRecs++
// Parse time, if exists.
if len(record) > 2 && record[2] != "" {
@ -239,54 +244,46 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowK
if err != nil {
return fmt.Errorf("invalid timestamp on row %d: %q", rnum, record[2])
}
bit.Timestamp = t.UnixNano()
if lastTime < batchRecs {
req.Timestamps = append(req.Timestamps, make([]int64, batchRecs-lastTime)...)
}
req.Timestamps = append(req.Timestamps, t.UnixNano())
lastTime = batchRecs
}
a = append(a, bit)
// If we've reached the buffer size then import bits.
if len(a) == cmd.BufferSize {
if err := cmd.importBits(ctx, useColumnKeys, useRowKeys, a); err != nil {
if batchRecs == cmd.BufferSize {
// pad timestamps out with 0s
if lastTime > 0 && lastTime < batchRecs {
req.Timestamps = append(req.Timestamps, make([]int64, batchRecs-lastTime)...)
}
if err := cmd.importBits(ctx, req); err != nil {
return err
}
a = a[:0]
req.ColumnIDs = req.ColumnIDs[:0]
req.RowIDs = req.RowIDs[:0]
req.ColumnKeys = req.ColumnKeys[:0]
req.RowKeys = req.RowKeys[:0]
req.Timestamps = req.Timestamps[:0]
lastTime = 0
batchRecs = 0
}
}
// If there are still bits in the buffer then flush them.
return cmd.importBits(ctx, useColumnKeys, useRowKeys, a)
if batchRecs == 0 {
return nil
}
if lastTime > 0 && lastTime < batchRecs {
req.Timestamps = append(req.Timestamps, make([]int64, batchRecs-lastTime)...)
}
return cmd.importBits(ctx, req)
}
// importBits sends batches of bits to the server.
func (cmd *ImportCommand) importBits(ctx context.Context, useColumnKeys, useRowKeys bool, bits []pilosa.Bit) error {
logger := log.New(cmd.Stderr, "", log.LstdFlags)
// If keys are used, all bits are sent to the primary translate store.
if useColumnKeys || useRowKeys {
logger.Printf("importing keys: n=%d", len(bits))
if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits, pilosa.OptImportOptionsClear(cmd.Clear)); err != nil {
return errors.Wrap(err, "importing keys")
}
return nil
}
// Group bits by shard.
logger.Printf("grouping %d bits", len(bits))
bitsByShard := http.Bits(bits).GroupByShard()
// Parse path into bits.
for shard, chunk := range bitsByShard {
if cmd.Sort {
sort.Sort(http.BitsByPos(chunk))
}
logger.Printf("importing shard: %d, n=%d", shard, len(chunk))
if err := cmd.client.Import(ctx, cmd.Index, cmd.Field, shard, chunk, pilosa.OptImportOptionsClear(cmd.Clear)); err != nil {
return errors.Wrap(err, "importing")
}
}
return nil
func (cmd *ImportCommand) importBits(ctx context.Context, req *pilosa.ImportRequest) error {
req.Shard = ^uint64(0)
return cmd.client.Import(ctx, nil, req, &pilosa.ImportOptions{Clear: cmd.Clear})
}
// bufferValues buffers slices of record identifiers and values to be imported as a batch.
@ -360,7 +357,7 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, useColumnKeys, parse
// If we've reached the buffer size then import the batch.
if len(req.ColumnKeys) == cmd.BufferSize || len(req.ColumnIDs) == cmd.BufferSize {
if err := cmd.client.ImportValue2(ctx, req, &pilosa.ImportOptions{}); err != nil {
if err := cmd.client.ImportValue(ctx, nil, req, &pilosa.ImportOptions{Clear: cmd.Clear}); err != nil {
return errors.Wrap(err, "importing values")
}
req.ColumnIDs = req.ColumnIDs[:0]
@ -371,7 +368,7 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, useColumnKeys, parse
}
// If there are still values in the buffer then flush them.
return errors.Wrap(cmd.client.ImportValue2(ctx, req, &pilosa.ImportOptions{}), "importing values")
return errors.Wrap(cmd.client.ImportValue(ctx, nil, req, &pilosa.ImportOptions{Clear: cmd.Clear}), "importing values")
}
func (cmd *ImportCommand) TLSHost() string {

8
executor_test.go
View file

@ -4131,13 +4131,11 @@ func TestExecutor_Execute_All(t *testing.T) {
m0 := c.GetNode(0)
// the request gets altered by the Import operation now...
reqs, err := req.Clone().ShardSplit()
if err != nil {
t.Fatalf("splitting request into shards: %v", err)
}
reqs := req.Clone().SortToShards()
qcx := m0.API.Txf().NewQcx()
for _, r := range reqs {
// we can ignore the key (which is the shard) because each req
// also got its internal key set.
if err := m0.API.Import(context.Background(), qcx, r); err != nil {
t.Fatal(err)
}

12
field.go
View file

@ -1483,17 +1483,7 @@ func (f *Field) Range(qcx *Qcx, name string, op pql.Token, predicate int64) (*Ro
}
// Import bulk imports data.
func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, shard uint64, opts ...ImportOption) (err0 error) {
// Set up import options.
options := &ImportOptions{}
for _, opt := range opts {
err := opt(options)
if err != nil {
return errors.Wrap(err, "applying option")
}
}
func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, shard uint64, options *ImportOptions) (err0 error) {
// Determine quantum if timestamps are set.
q := f.TimeQuantum()
if len(timestamps) > 0 {

1
go.mod
View file

@ -15,6 +15,7 @@ require (
github.com/davecgh/go-spew v1.1.1
github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect
github.com/dustin/go-humanize v1.0.0 // indirect
github.com/felixge/fgprof v0.9.1
github.com/fsnotify/fsnotify v1.4.9 // indirect
github.com/go-test/deep v1.0.7
github.com/gogo/protobuf v1.3.2

9
go.sum
View file

@ -47,6 +47,9 @@ github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx2
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa h1:OaNxuTZr7kxeODyLWsRMC+OD03aFUH+mW6r2d+MWa5Y=
@ -81,6 +84,8 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m
github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
github.com/felixge/fgprof v0.9.1 h1:E6FUJ2Mlv043ipLOCFqo8+cHo9MhQ203E2cdEK/isEs=
github.com/felixge/fgprof v0.9.1/go.mod h1:7/HK6JFtFaARhIljgP2IV8rJLIoHDoOYoUphsnGvqxE=
github.com/fogleman/gg v1.2.1-0.20190220221249-0403632d5b90/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzPa1k=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
@ -127,6 +132,8 @@ github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
github.com/google/pprof v0.0.0-20200615235658-03e1cf38a040 h1:i7RUpu0EybzQyQvPT7J3MmODs4+gPcHsD/pqW0uIYVo=
github.com/google/pprof v0.0.0-20200615235658-03e1cf38a040/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.1.4 h1:0ecGp3skIrHWPNGPJDaBIghfA6Sp7Ruo2Io8eLKzWm0=
@ -172,6 +179,7 @@ github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO
github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=
github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
github.com/improbable-eng/grpc-web v0.13.0 h1:7XqtaBWaOCH0cVGKHyvhtcuo6fgW32Y10yRKrDHFHOc=
github.com/improbable-eng/grpc-web v0.13.0/go.mod h1:6hRR09jOEG81ADP5wCQju1z71g6OL4eEvELdran/3cs=
github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
@ -441,6 +449,7 @@ golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191220142924-d4481acd189f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

57
handler.go
View file

@ -18,7 +18,7 @@ import (
"encoding/json"
"time"
"github.com/molecula/featurebase/v2/shardwidth"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/tracing"
"github.com/pkg/errors"
)
@ -307,40 +307,31 @@ func (ir *ImportRequest) Clone() *ImportRequest {
return newIR
}
// ShardSplit splits the request into a slice of import requests. It requires
// that the original request have all elements sorted, and already have
// column IDs, not column keys.
func (ir *ImportRequest) ShardSplit() ([]*ImportRequest, error) {
if ir == nil {
return nil, nil
// SortToShards takes an import request which has been translated, but may
// not be sorted, and turns it into a map from shard IDs to individual import
// requests. We don't sort the entries within each shard because the correct
// sorting depends on the field type and we don't want to deal with that
// here.
func (ir *ImportRequest) SortToShards() map[uint64]*ImportRequest {
// cheat: use ingest
fo := ingest.FieldOperation{
RecordIDs: ir.ColumnIDs,
Values: ir.RowIDs,
Signed: ir.Timestamps,
}
// fix shard
if len(ir.ColumnIDs) < 2 {
ir.Shard = ir.ColumnIDs[0] >> shardwidth.Exponent
return []*ImportRequest{ir}, nil
sharded := fo.SortToShards()
output := make(map[uint64]*ImportRequest, len(sharded))
for shard, shardOp := range sharded {
shardReq := *ir
shardReq.ColumnKeys = nil
shardReq.RowKeys = nil
shardReq.Shard = shard
shardReq.ColumnIDs = shardOp.RecordIDs
shardReq.RowIDs = shardOp.Values
shardReq.Timestamps = shardOp.Signed
output[shard] = &shardReq
}
shards, ends := shardwidth.FindShards(ir.ColumnIDs)
out := make([]*ImportRequest, len(shards))
prev := 0
for i, shard := range shards {
next := ends[i]
newIR := &ImportRequest{}
*newIR = *ir
newIR.ColumnIDs = ir.ColumnIDs[prev:next:next]
if ir.RowIDs != nil {
newIR.RowIDs = ir.RowIDs[prev:next:next]
}
if ir.RowKeys != nil {
newIR.RowKeys = ir.RowKeys[prev:next:next]
}
if ir.Timestamps != nil {
newIR.Timestamps = ir.Timestamps[prev:next:next]
}
newIR.Shard = shard
out[i] = newIR
prev = next
}
return out, nil
return output
}
// ValidateWithTimestamp ensures that the payload of the request is valid.

379
http/client.go
View file

@ -46,9 +46,13 @@ type InternalClient struct {
// The client to use for HTTP communication.
httpClient *http.Client
// the local node's API, used for operations that we can short-circuit that way
api *pilosa.API
}
// NewInternalClient returns a new instance of InternalClient to connect to host.
// If api is non-nil, the client uses it for some same-host operations instead
// of going through http.
func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, error) {
if host == "" {
return nil, pilosa.ErrHostRequired
@ -542,47 +546,6 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index str
return qresp, nil
}
// Import bulk imports bits for a single shard to a host.
func (c *InternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []pilosa.Bit, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
return pilosa.ErrFieldRequired
}
// Set up import options.
options := &pilosa.ImportOptions{}
for _, opt := range opts {
err := opt(options)
if err != nil {
return errors.Wrap(err, "applying option")
}
}
buf, err := c.marshalImportPayload(index, field, shard, bits)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
// Retrieve a list of nodes that own the shard.
nodes, err := c.FragmentNodes(ctx, index, shard)
if err != nil {
return fmt.Errorf("shard nodes: %s", err)
}
// Import to each node.
for _, node := range nodes {
if err := c.importNode(ctx, node, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
}
}
return nil
}
func getPrimaryNode(nodes []*topology.Node) *topology.Node {
for _, node := range nodes {
if node.IsPrimary {
@ -592,57 +555,6 @@ func getPrimaryNode(nodes []*topology.Node) *topology.Node {
return nil
}
// ImportK bulk imports bits specified by string keys to a host.
func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits []pilosa.Bit, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportK")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
return pilosa.ErrFieldRequired
}
// Set up import options.
options := &pilosa.ImportOptions{}
for _, opt := range opts {
err := opt(options)
if err != nil {
return errors.Wrap(err, "applying option")
}
}
buf, err := c.marshalImportPayload(index, field, 0, bits)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
// Get the primary node; all bits are sent to the
// primary translate store (i.e. primary).
// TODO... is that right^^?
// RESPONSE: It looks like in ctl/import.go, we could change the
// logic in ImportCommand.importBits() to only use ImportK
// when useRowKeys = true. It's no longer necessary to
// send column key translations to the primary (although
// it should still work). As far as I know, the only thing
// that uses ImportK is the pilosa import sub-command.
nodes, err := c.Nodes(ctx)
if err != nil {
return fmt.Errorf("getting nodes: %s", err)
}
coord := getPrimaryNode(nodes)
if coord == nil {
return fmt.Errorf("could not find the primary node")
}
// Import to node.
if err := c.importNode(ctx, coord, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err)
}
return nil
}
func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options pilosa.IndexOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureIndex")
defer span.Finish()
@ -670,32 +582,6 @@ func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName s
return err
}
// marshalImportPayload marshalls the import parameters into a protobuf byte slice.
func (c *InternalClient) marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
rowIDs := Bits(bits).RowIDs()
rowKeys := Bits(bits).RowKeys()
columnIDs := Bits(bits).ColumnIDs()
columnKeys := Bits(bits).ColumnKeys()
timestamps := Bits(bits).Timestamps()
// Marshal data to protobuf.
buf, err := c.serializer.Marshal(&pilosa.ImportRequest{
Index: index,
Field: field,
Shard: shard,
RowIDs: rowIDs,
RowKeys: rowKeys,
ColumnIDs: columnIDs,
ColumnKeys: columnKeys,
Timestamps: timestamps,
})
if err != nil {
return nil, fmt.Errorf("marshal import request: %s", err)
}
return buf, nil
}
// importNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, index, field string, buf []byte, opts *pilosa.ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.importNode")
@ -747,134 +633,163 @@ func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, in
return nil
}
// ImportValue bulk imports field values for a single shard to a host.
func (c *InternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValue")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
return pilosa.ErrFieldRequired
}
// Set up import options.
options := &pilosa.ImportOptions{}
for _, opt := range opts {
err := opt(options)
// importHelper is an experiment to see whether SonarCloud's code duplication
// complaints make sense to address in this context, given the impracticality
// of refactoring ImportRequest/ImportValueRequest right now. The process
// function exists because we would use either api.ImportValueWithTx or
// api.ImportWithTx, passing it the actual underlying-type of req, but doing
// that in here with a type switch seems messy. Similarly, index/field/shard
// exist because we can't access those members of the two slightly different
// structs.
func (c *InternalClient) importHelper(ctx context.Context, req pilosa.Message, process func() error, index string, field string, shard uint64, options *pilosa.ImportOptions) error {
// If we don't actually know what shards we're sending to, and we have
// a local API and a qcx, we'll have a process function that uses the local
// API. Otherwise, even if we have an API
var nodes []*topology.Node
var err error
if shard != ^uint64(0) {
// we need a list of nodes specific to this shard.
nodes, err = c.FragmentNodes(ctx, index, shard)
if err != nil {
return errors.Wrap(err, "applying option")
return errors.Errorf("shard nodes: %s", err)
}
} else {
// We don't know what shard to use, any shard is fine, local host
// is better if available.
if process != nil {
// skip the HTTP round-trip if we can.
err = process()
// Note that Wrap(nil, ...) is still nil.
return errors.Wrap(err, "local import")
}
// get the complete list of nodes, so if we have an API, we can
// pick our local node and probably avoid actually sending the http
// request over the wire, even though we still have to go through
// the http interface.
nodes, err = c.Nodes(ctx)
}
// "us" is a usable local node if any, "them" is every node that we need
// to process which isn't that node. We start out with us == nil and
// them = the whole set of nodes.
var us *topology.Node
var them []*topology.Node = nodes
// If we have an API, we know what node we are. Even if we don't have
// a Qcx, we still care, because looping back to the local node will
// be faster than going to another node.
if c.api != nil {
myID := c.api.NodeID()
for i, node := range nodes {
if myID == node.ID {
// swap our node into the first position
nodes[i], nodes[0] = nodes[0], nodes[i]
// If we have a qcx, we'll treat our node even MORE
// specially.
if process != nil {
us, them = nodes[0], nodes[1:]
}
break
}
}
}
buf, err := c.marshalImportValuePayload(index, field, shard, vals)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
// If we had a valid API and Qcx, and shard was ^0, we'd have handled
// it previously. So if we get here, we don't have both a Qcx and an API,
// but we might have an API, in which case we'll have shuffled our node
// into the first position. Otherwise we're just taking whatever the first
// node is.
if shard == ^uint64(0) {
them = them[:1]
}
// Retrieve a list of nodes that own the shard.
nodes, err := c.FragmentNodes(ctx, index, shard)
if err != nil {
return fmt.Errorf("shard nodes: %s", err)
}
// Import to each node.
for _, node := range nodes {
if err := c.importNode(ctx, node, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
}
}
return nil
}
// ImportValue2 is a simplified ImportValue method which just uses the
// ImportValueRequest instead of splitting up ImportValue and
// ImportValueK... it also supports importing float values. The idea
// being that (assuming it works) this will become the default (and be
// renamed) for 2.0, and we can deprecate the other methods.
func (c *InternalClient) ImportValue2(ctx context.Context, req *pilosa.ImportValueRequest, options *pilosa.ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.NewImportValue")
defer span.Finish()
buf, err := c.serializer.Marshal(req)
if err != nil {
return errors.Errorf("marshal import request: %s", err)
}
// Retrieve a list of nodes that own the shard.
nodes, err := c.FragmentNodes(ctx, req.Index, req.Shard)
if err != nil {
return errors.Errorf("shard nodes: %s", err)
}
// Import to each node.
for _, node := range nodes {
if err := c.importNode(ctx, node, req.Index, req.Field, buf, options); err != nil {
return errors.Errorf("import node: host=%s, err=%s", node.URI, err)
}
}
return nil
}
// ImportValueK bulk imports keyed field values to a host.
func (c *InternalClient) ImportValueK(ctx context.Context, index, field string, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValueK")
defer span.Finish()
buf, err := c.marshalImportValuePayload(index, field, 0, vals)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
// Set up import options.
options := &pilosa.ImportOptions{}
for _, opt := range opts {
err := opt(options)
// We handle remote nodes first, for two distinct reasons. One is that
// the local API ImportWithTx is allowed to modify its inputs, and if we
// ran that before serializing, we'd get corrupt data serialized.
// The other is that if we were to hold a write lock that started with
// that import and ended when we hit the end of our Qcx, we wouldn't want
// to hold it during all our requests to the remote nodes.
if len(them) > 0 {
buf, err := c.serializer.Marshal(req)
if err != nil {
return errors.Wrap(err, "applying option")
return errors.Errorf("marshal import request: %s", err)
}
// We process remote nodes first so we won't be actually holding our
// write lock yet, in theory. This doesn't actually matter yet, but is
// helpful for future planned refactoring.
for _, node := range them {
if err = c.importNode(ctx, node, index, field, buf, options); err != nil {
return errors.Wrap(err, "remote import")
}
}
}
// Get the primary node; all bits are sent to the
// primary translate store.
nodes, err := c.Nodes(ctx)
if err != nil {
return fmt.Errorf("getting nodes: %s", err)
// Write to the local node if we have one.
if us != nil {
// WARNING: ImportWithTx can alter its inputs. However, we can
// only ever do this once, and if we're going to need a marshalled
// form, we already made it.
if err = process(); err != nil {
return errors.Wrap(err, "local import after remote imports")
}
}
coord := getPrimaryNode(nodes)
if coord == nil {
return fmt.Errorf("could not find the primary node")
}
// Import to node.
if err := c.importNode(ctx, coord, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err)
}
return nil
}
// marshalImportValuePayload marshalls the import parameters into a protobuf byte slice.
func (c *InternalClient) marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
columnIDs := FieldValues(vals).ColumnIDs()
columnKeys := FieldValues(vals).ColumnKeys()
values := FieldValues(vals).Values()
// Import imports values using an ImportRequest, whether or not it's keyed.
// It may modify the contents of req.
//
// If a request comes in with Shard -1, it will be sent to only one node,
// which will translate if necessary, split into shards, and loop back
// through this for each sub-request. If a request uses record keys,
// it will be set to use shard = -1 unconditionally, because we know
// that it has to be translated and possibly reshuffled. Value keys
// don't override the shard.
//
// If we get a non-nil qcx, and have an associated API, we'll use that API
// directly for the local shard.
func (c *InternalClient) Import(ctx context.Context, qcx *pilosa.Qcx, req *pilosa.ImportRequest, options *pilosa.ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import")
defer span.Finish()
// Marshal data to protobuf.
buf, err := c.serializer.Marshal(&pilosa.ImportValueRequest{
Index: index,
Field: field,
Shard: shard,
ColumnIDs: columnIDs,
ColumnKeys: columnKeys,
Values: values,
})
if err != nil {
return nil, fmt.Errorf("marshal import request: %s", err)
if req.ColumnKeys != nil {
req.Shard = ^uint64(0)
}
return buf, nil
var process func() error
if c.api != nil && qcx != nil {
process = func() error {
return c.api.ImportWithTx(ctx, qcx, req, options)
}
}
return c.importHelper(ctx, req, process, req.Index, req.Field, req.Shard, options)
}
// ImportValue imports values using an ImportValueRequest, whether or not it's
// keyed. It may modify the contents of req.
//
// If a request comes in with Shard -1, it will be sent to only one node,
// which will translate if necessary, split into shards, and loop back
// through this for each sub-request. If a request uses record keys,
// it will be set to use shard = -1 unconditionally, because we know
// that it has to be translated and possibly reshuffled. Value keys
// don't override the shard.
//
// If we get a non-nil qcx, and have an associated API, we'll use that API
// directly for the local shard.
func (c *InternalClient) ImportValue(ctx context.Context, qcx *pilosa.Qcx, req *pilosa.ImportValueRequest, options *pilosa.ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import")
defer span.Finish()
if req.ColumnKeys != nil {
req.Shard = ^uint64(0)
}
var process func() error
if c.api != nil && qcx != nil {
process = func() error {
return c.api.ImportValueWithTx(ctx, qcx, req, options)
}
}
return c.importHelper(ctx, req, process, req.Index, req.Field, req.Shard, options)
}
// ImportRoaring does fast import of raw bits in roaring format (pilosa or
@ -2336,3 +2251,7 @@ func (c *InternalClient) PartitionNodes(ctx context.Context, partitionID int) ([
}
return a, nil
}
func (c *InternalClient) SetInternalAPI(api *pilosa.API) {
c.api = api
}

349
http/client_test.go
View file

@ -202,7 +202,6 @@ func TestClient_Export(t *testing.T) {
cmd.MustCreateField(t, "unkeyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000))
c := MustNewClient(host, http.GetHTTPClient(nil))
data := []pilosa.Bit{
{RowID: 1, ColumnID: 100, RowKey: "row1", ColumnKey: "col100"},
{RowID: 1, ColumnID: 101, RowKey: "row1", ColumnKey: "col101"},
@ -367,49 +366,155 @@ func TestClient_Export(t *testing.T) {
// Ensure client can bulk import data.
func TestClient_Import(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
// Need a cluster to verify hitting multiple nodes
cluster := test.MustRunCluster(t, 3)
defer cluster.Close()
cmd := cluster.GetNode(0)
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
api := cmd.API
// Load bitmap into cache to ensure cache gets updated.
hldr.SetBit("i", "f", 1, 0) // set a bit so the view gets created.
hldr.Row("i", "f", 0)
cmd.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true})
cmd.MustCreateIndex(t, "unkeyed", pilosa.IndexOptions{Keys: false})
cmd.MustCreateField(t, "keyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys())
cmd.MustCreateField(t, "keyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0))
cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys())
cmd.MustCreateField(t, "unkeyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeNone, 0))
indexes := map[bool]string{true: "keyed", false: "unkeyed"}
fields := map[bool]string{true: "keyedf", false: "unkeyedf"}
recKeys := []string{"rec-a", "rec-b", "rec-c"}
valueKeys := []string{"val-a", "val-b", "val-c"}
recIDs := []uint64{0, 3, 7}
valueIDs := []uint64{0, 3, 7}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.Import(context.Background(), "i", "f", 0, []pilosa.Bit{
{RowID: 0, ColumnID: 1},
{RowID: 0, ColumnID: 5},
{RowID: 200, ColumnID: 6},
}); err != nil {
t.Fatal(err)
// set API to point at the local node
c.SetInternalAPI(cmd.API)
checkResults := func(results pilosa.QueryResponse, keyed bool, maxN int) {
for i, r := range results.Results {
row, ok := r.(*pilosa.Row)
if !ok {
t.Fatalf("expected row, got %T", r)
}
if i >= maxN {
// we cleared these, so they should be empty
if keyed {
vals := row.Keys
if len(vals) != 0 {
t.Fatalf("expected empty row, got %d result(s) back, first result %q",
len(vals), vals[0])
}
} else {
vals := row.Columns()
if len(vals) != 0 {
t.Fatalf("expected empty row, got %d result(s) back, first result %d",
len(vals), vals[0])
}
}
return
}
if keyed {
vals := row.Keys
if len(vals) != 1 {
t.Fatalf("expected one result, didn't get it")
}
if vals[0] != recKeys[i] {
t.Fatalf("expected %q, got %q", recKeys[i], vals[0])
}
} else {
vals := row.Columns()
if len(vals) != 1 {
t.Fatalf("expected one result, didn't get it")
}
if vals[0] != recIDs[i] {
t.Fatalf("expected %d, got %d", recIDs[i], vals[0])
}
}
}
}
// Verify data.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 5}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{6}) {
t.Fatalf("unexpected columns: %+v", a)
}
for keyed, indexName := range indexes {
for _, fieldName := range fields {
req := pilosa.ImportRequest{
Index: indexName,
Field: fieldName,
}
if indexName == "keyed" {
req.ColumnKeys = recKeys
req.Shard = ^uint64(0)
} else {
req.ColumnIDs = recIDs
req.Shard = 0
}
if fieldName == "keyedf" {
req.RowKeys = valueKeys
} else {
req.RowIDs = valueIDs
}
// clone request because some imports will modify their input parameters
if err := c.Import(context.Background(), nil, req.Clone(), &pilosa.ImportOptions{}); err != nil {
t.Fatalf("%s/%s: %v",
indexName, fieldName, err)
}
// Clear some data.
if err := c.Import(context.Background(), "i", "f", 0, []pilosa.Bit{
{RowID: 0, ColumnID: 5},
{RowID: 200, ColumnID: 6},
}, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Now do a query to see whether it worked...
var pql string
if fieldName == "keyedf" {
pql = `Row(keyedf="val-a") Row(keyedf="val-b") Row(keyedf="val-c")`
} else {
pql = `Row(unkeyedf=0) Row(unkeyedf=3) Row(unkeyedf=7)`
}
results, err := api.Query(context.Background(), &pilosa.QueryRequest{Index: indexName, Query: pql})
if err != nil {
t.Fatal(err)
}
checkResults(results, keyed, 3)
// Verify data.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{}) {
t.Fatalf("unexpected columns: %+v", a)
// Now clear a bit...
req = pilosa.ImportRequest{
Index: indexName,
Field: fieldName,
}
if indexName == "keyed" {
req.ColumnKeys = recKeys[2:]
req.Shard = ^uint64(0)
} else {
req.ColumnIDs = recIDs[2:]
req.Shard = 0
}
if fieldName == "keyedf" {
req.RowKeys = valueKeys[2:]
} else {
req.RowIDs = valueIDs[2:]
}
// do a clear. also, do the clear with a Qcx.
func() {
// inner function so the deferred abort isn't delayed a lot
qcx := api.Txf().NewQcx()
defer qcx.Abort()
if err := c.Import(context.Background(), qcx, req.Clone(), &pilosa.ImportOptions{Clear: true}); err != nil {
t.Fatalf("%s/%s: %v",
indexName, fieldName, err)
}
if err := qcx.Finish(); err != nil {
t.Fatalf("committing write: %v", err)
}
}()
// Now do a query to see whether it worked...
if fieldName == "keyedf" {
pql = `Row(keyedf="val-a") Row(keyedf="val-b") Row(keyedf="val-c")`
} else {
pql = `Row(unkeyedf=0) Row(unkeyedf=3) Row(unkeyedf=7)`
}
results, err = api.Query(context.Background(), &pilosa.QueryRequest{Index: indexName, Query: pql})
if err != nil {
t.Fatal(err)
}
checkResults(results, keyed, 2)
}
}
}
@ -547,15 +652,17 @@ func TestClient_ImportRoaring_MultiView(t *testing.T) {
}
defer cluster.Close()
_, err = cluster.GetNode(0).API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
api := cluster.GetNode(0).API
_, err = api.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = cluster.GetNode(0).API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100))
_, err = api.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100))
if err != nil {
t.Fatalf("creating field: %v", err)
}
_, err = cluster.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "Set(0, f=1)"})
_, err = api.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "Set(0, f=1)"})
if err != nil {
t.Fatalf("querying: %v", err)
}
@ -588,16 +695,21 @@ func TestClient_ImportKeys(t *testing.T) {
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
baseReq := &pilosa.ImportRequest{
Index: "keyed",
Field: "keyedf",
ColumnKeys: []string{"eve", "alice", "bob", "eve", "alice", "eve"},
ColumnIDs: []uint64{1, 2, 3, 1, 2, 1},
RowKeys: []string{"green", "green", "green", "blue", "blue", "purple"},
RowIDs: []uint64{1, 1, 1, 2, 2, 3},
}
t.Run("Import keyed,keyed", func(t *testing.T) {
if err := c.Import(context.Background(), "keyed", "keyedf", 0, []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
req := baseReq.Clone()
req.Index = "keyed"
req.Field = "keyedf"
req.ColumnIDs, req.RowIDs = nil, nil
if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
resp := cmd.QueryAPI(t, &pilosa.QueryRequest{
@ -616,14 +728,11 @@ func TestClient_ImportKeys(t *testing.T) {
})
t.Run("Import keyed,unkeyedf", func(t *testing.T) {
if err := c.Import(context.Background(), "keyed", "unkeyedf", 0, []pilosa.Bit{
{RowID: 1, ColumnKey: "eve"},
{RowID: 1, ColumnKey: "alice"},
{RowID: 1, ColumnKey: "bob"},
{RowID: 2, ColumnKey: "eve"},
{RowID: 2, ColumnKey: "alice"},
{RowID: 3, ColumnKey: "eve"},
}); err != nil {
req := baseReq.Clone()
req.Index = "keyed"
req.Field = "unkeyedf"
req.ColumnIDs, req.RowKeys = nil, nil
if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
resp := cmd.QueryAPI(t, &pilosa.QueryRequest{
@ -642,14 +751,11 @@ func TestClient_ImportKeys(t *testing.T) {
})
t.Run("Import unkeyed,keyed", func(t *testing.T) {
if err := c.Import(context.Background(), "unkeyed", "keyedf", 0, []pilosa.Bit{
{RowKey: "green", ColumnID: 1},
{RowKey: "green", ColumnID: 2},
{RowKey: "green", ColumnID: 3},
{RowKey: "blue", ColumnID: 1},
{RowKey: "blue", ColumnID: 2},
{RowKey: "purple", ColumnID: 1},
}); err != nil {
req := baseReq.Clone()
req.Index = "unkeyed"
req.Field = "keyedf"
req.ColumnKeys, req.RowIDs = nil, nil
if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
resp := cmd.QueryAPI(t, &pilosa.QueryRequest{
@ -686,14 +792,13 @@ func TestClient_ImportKeys(t *testing.T) {
// Import to node0.
t.Run("Import node0", func(t *testing.T) {
if err := c0.ImportK(context.Background(), "keyed", "keyedf0", []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
req := &pilosa.ImportRequest{
Index: "keyed",
Field: "keyedf0",
ColumnKeys: []string{"eve", "alice", "bob", "eve", "alice", "eve"},
RowKeys: []string{"green", "green", "green", "blue", "blue", "purple"},
}
if err := c0.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
resp := cmd0.QueryAPI(t, &pilosa.QueryRequest{
@ -713,14 +818,13 @@ func TestClient_ImportKeys(t *testing.T) {
// Import to node1 (ensure import is routed to primary for translation).
t.Run("Import node1", func(t *testing.T) {
if err := c1.ImportK(context.Background(), "keyed", "keyedf1", []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
req := &pilosa.ImportRequest{
Index: "keyed",
Field: "keyedf1",
ColumnKeys: []string{"eve", "alice", "bob", "eve", "alice", "eve"},
RowKeys: []string{"green", "green", "green", "blue", "blue", "purple"},
}
if err := c1.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
@ -762,11 +866,13 @@ func TestClient_ImportKeys(t *testing.T) {
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnKey: "col1", Value: -10},
{ColumnKey: "col2", Value: 20},
{ColumnKey: "col3", Value: 40},
}); err != nil {
req := &pilosa.ImportValueRequest{
Index: "i",
Field: "f",
ColumnKeys: []string{"col1", "col2", "col3"},
Values: []int64{-10, 20, 40},
}
if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
@ -786,9 +892,13 @@ func TestClient_ImportKeys(t *testing.T) {
}
// Clear data.
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnKey: "col2", Value: 20},
}, pilosa.OptImportOptionsClear(true)); err != nil {
req = &pilosa.ImportValueRequest{
Index: "i",
Field: "f",
ColumnKeys: []string{"col2"},
Values: []int64{20},
}
if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{Clear: true}); err != nil {
t.Fatal(err)
}
@ -835,9 +945,13 @@ func TestClient_ImportIDs(t *testing.T) {
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{
{ColumnID: 2, Value: 1},
}); err != nil {
req := &pilosa.ImportValueRequest{
Index: idxName,
Field: fldName,
ColumnIDs: []uint64{2},
Values: []int64{1},
}
if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
@ -857,9 +971,13 @@ func TestClient_ImportIDs(t *testing.T) {
}
// Send import request.
if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{
{ColumnID: 1000, Value: 1},
}); err != nil {
req = &pilosa.ImportValueRequest{
Index: idxName,
Field: fldName,
ColumnIDs: []uint64{1000},
Values: []int64{1},
}
if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
@ -895,11 +1013,13 @@ func TestClient_ImportValue(t *testing.T) {
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnID: 1, Value: -10},
{ColumnID: 2, Value: 20},
{ColumnID: 3, Value: 40},
}); err != nil {
req := &pilosa.ImportValueRequest{
Index: "i",
Field: "f",
ColumnIDs: []uint64{1, 2, 3},
Values: []int64{-10, 20, 40},
}
if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
@ -922,10 +1042,13 @@ func TestClient_ImportValue(t *testing.T) {
}
// Send import request.
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnID: 1, Value: -10},
{ColumnID: 3, Value: 40},
}, pilosa.OptImportOptionsClear(true)); err != nil {
req = &pilosa.ImportValueRequest{
Index: "i",
Field: "f",
ColumnIDs: []uint64{1, 3},
Values: []int64{-10, 40},
}
if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{Clear: true}); err != nil {
t.Fatal(err)
}
@ -969,11 +1092,13 @@ func TestClient_ImportExistence(t *testing.T) {
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.Import(context.Background(), idxName, fldName, 0, []pilosa.Bit{
{RowID: 0, ColumnID: 1},
{RowID: 0, ColumnID: 5},
{RowID: 200, ColumnID: 6},
}); err != nil {
req := &pilosa.ImportRequest{
Index: "iset",
Field: "fset",
ColumnIDs: []uint64{1, 5, 6},
RowIDs: []uint64{0, 0, 200},
}
if err := c.Import(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
@ -1003,11 +1128,13 @@ func TestClient_ImportExistence(t *testing.T) {
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{
{ColumnID: 1, Value: -10},
{ColumnID: 2, Value: 20},
{ColumnID: 3, Value: 40},
}); err != nil {
req := &pilosa.ImportValueRequest{
Index: "iint",
Field: "fint",
ColumnIDs: []uint64{1, 2, 3},
Values: []int64{-10, 20, 40},
}
if err := c.ImportValue(context.Background(), nil, req, &pilosa.ImportOptions{}); err != nil {
t.Fatal(err)
}
@ -1087,7 +1214,7 @@ func TestClient_CreateDecimalField(t *testing.T) {
t.Fatalf("expected Scale 1, got: %+v", fld.Options())
}
err = c.ImportValue2(context.Background(), &pilosa.ImportValueRequest{Index: index, Field: field, ColumnIDs: []uint64{1, 2, 3}, Shard: 0, FloatValues: []float64{1.1, 2.2, 3.3}}, &pilosa.ImportOptions{})
err = c.ImportValue(context.Background(), nil, &pilosa.ImportValueRequest{Index: index, Field: field, ColumnIDs: []uint64{1, 2, 3}, Shard: 0, FloatValues: []float64{1.1, 2.2, 3.3}}, &pilosa.ImportOptions{})
if err != nil {
t.Fatalf("importing float values: %v", err)
}

2
http/handler.go
View file

@ -38,6 +38,7 @@ import (
"sync"
"time"
"github.com/felixge/fgprof"
"github.com/gorilla/handlers"
"github.com/gorilla/mux"
pilosa "github.com/molecula/featurebase/v2"
@ -373,6 +374,7 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST").Name("PostClusterResizeAbort")
router.HandleFunc("/cluster/resize/remove-node", handler.handlePostClusterResizeRemoveNode).Methods("POST").Name("PostClusterResizeRemoveNode")
router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET")
router.PathPrefix("/debug/fgprof").Handler(fgprof.Handler()).Methods("GET")
router.Handle("/debug/vars", expvar.Handler()).Methods("GET")
router.Handle("/metrics", promhttp.Handler())
router.HandleFunc("/metrics.json", handler.handleGetMetricsJSON).Methods("GET").Name("GetMetricsJSON")

4
ingest/op.go
View file

@ -243,8 +243,8 @@ func (f *FieldOperation) ByShard() ShardedFieldOperation {
func (f *FieldOperation) clone() *FieldOperation {
f2 := &FieldOperation{
RecordIDs: append([]uint64{}, f.RecordIDs...),
Values: append([]uint64{}, f.Values...),
Signed: append([]int64{}, f.Signed...),
Values: append(([]uint64)(nil), f.Values...),
Signed: append(([]int64)(nil), f.Signed...),
}
return f2
}

47
ingest/sort_test.go
View file

@ -14,6 +14,53 @@
package ingest
import (
"testing"
)
const sampleSize = 1000000
var sampleSortingData = createSampleFieldData()
func createSampleFieldData() *FieldOperation {
fo := &FieldOperation{}
fo.RecordIDs = make([]uint64, sampleSize)
fo.Values = make([]uint64, sampleSize)
fo.Signed = make([]int64, sampleSize)
for i := range fo.RecordIDs {
fo.RecordIDs[i] = (uint64(i) * 63 * 3456789) % 50000000
fo.Values[i] = (uint64(i) * 6) % 8
fo.Signed[i] = ((int64(i) * 17) % 15) - 8
}
return fo
}
func benchmarkOneSort(b *testing.B, fo *FieldOperation) {
for i := 0; i < b.N; i++ {
b.StopTimer()
sortable := fo.clone()
b.StartTimer()
_ = sortable.SortToShards()
}
}
func BenchmarkSortFieldOp(b *testing.B) {
b.Run("full", func(b *testing.B) {
f2 := *sampleSortingData
benchmarkOneSort(b, &f2)
})
b.Run("nosign", func(b *testing.B) {
f2 := *sampleSortingData
f2.Signed = nil
benchmarkOneSort(b, &f2)
})
b.Run("signonly", func(b *testing.B) {
f2 := *sampleSortingData
f2.Values = nil
benchmarkOneSort(b, &f2)
})
}
// func BenchmarkSort64(b *testing.B) {
// gen := func(n, width uint64) func() []uint64 {
// var data []uint64

16
internal/clustertests/cluster_test.go
View file

@ -53,13 +53,19 @@ func TestClusterStuff(t *testing.T) {
t.Fatalf("creating field: %v", err)
}
data := make([]pilosa.Bit, 10)
req := &pilosa.ImportRequest{
Index: "testidx",
Field: "testf",
}
req.ColumnIDs = make([]uint64, 10)
req.RowIDs = make([]uint64, 10)
for i := 0; i < 1000; i++ {
data[i%10].RowID = 0
data[i%10].ColumnID = uint64((i/10)*pilosa.ShardWidth + i%10)
shard := uint64(i / 10)
req.RowIDs[i%10] = 0
req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10)
req.Shard = uint64(i / 10)
if i%10 == 9 {
err = cli1.Import(context.Background(), "testidx", "testf", shard, data)
err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{})
if err != nil {
t.Fatalf("importing: %v", err)
}

15
nfpm.yaml Normal file
View file

@ -0,0 +1,15 @@
name: "featurebase"
arch: "${GOARCH}"
platform: "${GOOS}"
version: "${VERSION}"
section: "default"
priority: "extra"
maintainer: "Molecula Corp. <info@molecula.com>"
description: "FeatureBase is a feature extraction and storage technology that enables real-time analytics."
vendor: "Molecula"
homepage: "https://molecula.com"
contents:
- src: ./featurebase
dst: /usr/local/bin/featurebase
- dst: /etc/featurebase.conf
type: ghost

4
server.go
View file

@ -531,6 +531,10 @@ func (s *Server) GRPCURI() pnet.URI {
return s.grpcURI
}
func (s *Server) SetAPI(api *API) {
s.defaultClient.SetInternalAPI(api)
}
// UpAndDown brings the server up minimally and shuts it down
// again; basically, it exists for testing holder open and close.
func (s *Server) UpAndDown() error {

10
server/config_internal_test.go
View file

@ -35,7 +35,10 @@ func TestConfig_validateAddrs(t *testing.T) {
}
hostAddr, err := lookupAddr(context.Background(), net.DefaultResolver, hostname)
if err != nil {
t.Fatal(err)
if !strings.HasSuffix(hostname, ".local") {
t.Fatal(err)
}
hostAddr = outboundAddr
}
if strings.Contains(hostAddr, ":") {
hostAddr = "[" + hostAddr + "]"
@ -164,7 +167,10 @@ func TestConfig_validateAddrsGRPC(t *testing.T) {
}
hostAddr, err := lookupAddr(context.Background(), net.DefaultResolver, hostname)
if err != nil {
t.Fatal(err)
if !strings.HasSuffix(hostname, ".local") {
t.Fatal(err)
}
hostAddr = outboundAddr
}
if strings.Contains(hostAddr, ":") {
hostAddr = "[" + hostAddr + "]"

2
server/server.go
View file

@ -516,6 +516,8 @@ func (m *Command) SetupServer() error {
if err != nil {
return errors.Wrap(err, "new api")
}
// Tell server about its new API, which its client will need.
m.Server.SetAPI(m.API)
m.grpcServer, err = NewGRPCServer(
OptGRPCServerAPI(m.API),

2
server/server_test.go
View file

@ -67,6 +67,7 @@ func TestMain_Set_Quick(t *testing.T) {
// Create client.
client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil))
client.SetInternalAPI(m.API)
if err != nil {
t.Fatal(err)
}
@ -831,6 +832,7 @@ func TestQueryingWithQuotesAndStuff(t *testing.T) {
defer m.Close()
client, err := http.NewInternalClient(m.API.Node().URI.HostPort(), http.GetHTTPClient(nil))
client.SetInternalAPI(m.API)
if err != nil {
t.Fatal(err)
}