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

Merge branch 'master' into cluster-tests

Browse source
This commit is contained in:
Matt Jaffee 2018-10-26 16:14:00 -05:00
commit ec08930c30
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
24 changed files with 3591 additions and 327 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

75
api.go
View file

@ -239,6 +239,17 @@ func (api *API) Field(_ context.Context, indexName, fieldName string) (*Field, e
return field, nil
}
func setUpImportOptions(opts ...ImportOption) (*ImportOptions, error) {
options := &ImportOptions{}
for _, opt := range opts {
err := opt(options)
if err != nil {
return nil, errors.Wrap(err, "applying option")
}
}
return options, nil
}
// ImportRoaring is a low level interface for importing data to Pilosa when
// extremely high throughput is desired. The data must be encoded in a
// particular way which may be unintuitive (discussed below). The data is merged
@ -256,10 +267,17 @@ func (api *API) Field(_ context.Context, indexName, fieldName string) (*Field, e
// (shard*ShardWidth)+(i%ShardWidth). That is to say that "data" represents all
// of the rows in this shard of this field concatenated together in one long
// bitmap.
func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, remote bool, data []byte) (err error) {
func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, remote bool, data []byte, opts ...ImportOption) (err error) {
if err = api.validate(apiField); err != nil {
return errors.Wrap(err, "validating api method")
}
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
nodes := api.cluster.shardNodes(indexName, shard)
var eg errgroup.Group
@ -280,14 +298,14 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
d2 := make([]byte, len(data))
copy(d2, data)
eg.Go(func() error {
return field.importRoaring(d2, shard)
return field.importRoaring(d2, shard, options.Clear)
})
go func(node *Node) {
}(node)
} else if !remote { // if remote == true we don't forward to other nodes
// forward it on
eg.Go(func() error {
return api.server.defaultClient.ImportRoaring(ctx, &node.URI, indexName, fieldName, shard, true, data)
return api.server.defaultClient.ImportRoaring(ctx, &node.URI, indexName, fieldName, shard, true, data, opts...)
})
}
}
@ -670,12 +688,33 @@ func (api *API) FieldAttrDiff(_ context.Context, indexName string, fieldName str
return attrs, nil
}
// ImportOptions holds the options for the API.Import method.
type ImportOptions struct {
Clear bool
}
// ImportOption is a functional option type for API.Import.
type ImportOption func(*ImportOptions) error
func OptImportOptionsClear(c bool) ImportOption {
return func(o *ImportOptions) error {
o.Clear = c
return nil
}
}
// Import bulk imports data into a particular index,field,shard.
func (api *API) Import(_ context.Context, req *ImportRequest) error {
func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOption) error {
if err := api.validate(apiImport); err != nil {
return errors.Wrap(err, "validating api method")
}
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
index := api.holder.Index(req.Index)
if index == nil {
return newNotFoundError(ErrIndexNotFound)
@ -717,13 +756,15 @@ func (api *API) Import(_ context.Context, req *ImportRequest) error {
}
// Import columnIDs into existence field.
if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
api.server.logger.Printf("import existence 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 existence columns")
if !options.Clear {
if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
api.server.logger.Printf("import existence 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 existence columns")
}
}
// Import into fragment.
err = field.Import(req.RowIDs, req.ColumnIDs, timestamps)
err = field.Import(req.RowIDs, req.ColumnIDs, timestamps, opts...)
if err != nil {
api.server.logger.Printf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
}
@ -731,11 +772,17 @@ func (api *API) Import(_ context.Context, req *ImportRequest) error {
}
// ImportValue bulk imports values into a particular field.
func (api *API) ImportValue(_ context.Context, req *ImportValueRequest) error {
func (api *API) ImportValue(_ context.Context, req *ImportValueRequest, opts ...ImportOption) error {
if err := api.validate(apiImportValue); err != nil {
return errors.Wrap(err, "validating api method")
}
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
return errors.Wrap(err, "setting up import options")
}
index := api.holder.Index(req.Index)
if index == nil {
return newNotFoundError(ErrIndexNotFound)
@ -757,13 +804,15 @@ func (api *API) ImportValue(_ context.Context, req *ImportValueRequest) error {
}
// Import columnIDs into existence field.
if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
api.server.logger.Printf("import existence 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 existence columns")
if !options.Clear {
if err := importExistenceColumns(index, req.ColumnIDs); err != nil {
api.server.logger.Printf("import existence 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 existence columns")
}
}
// Import into fragment.
err = field.importValue(req.ColumnIDs, req.Values)
err = field.importValue(req.ColumnIDs, req.Values, options)
if err != nil {
api.server.logger.Printf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
}

20
client.go
View file

@ -37,13 +37,13 @@ type InternalClient interface {
Nodes(ctx context.Context) ([]*Node, error)
Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error)
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error
ImportK(ctx context.Context, index, field string, bits []Bit) 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
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) error
ImportValueK(ctx context.Context, index, field string, vals []FieldValue) 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
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
@ -53,7 +53,7 @@ type InternalClient interface {
RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
SendMessage(ctx context.Context, uri *URI, msg []byte) error
RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error)
ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, data []byte) error
ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, data []byte, opts ...ImportOption) error
}
//===============
@ -103,13 +103,13 @@ func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest
func (n nopInternalClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n nopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error {
func (n nopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error {
return nil
}
func (n nopInternalClient) ImportK(ctx context.Context, index, field string, bits []Bit) error {
func (n nopInternalClient) ImportK(ctx context.Context, index, field string, bits []Bit, opts ...ImportOption) error {
return nil
}
func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, data []byte) error {
func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, data []byte, opts ...ImportOption) error {
return nil
}
func (n nopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error {
@ -121,10 +121,10 @@ 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) error {
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) error {
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 {

1
cmd/import.go
View file

@ -63,6 +63,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
flags.IntVarP(&Importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.")
flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.")
flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.")
flags.BoolVarP(&Importer.Clear, "clear", "", false, "Clear the data provided in the import.")
ctl.SetTLSConfig(flags, &Importer.TLS.CertificatePath, &Importer.TLS.CertificateKeyPath, &Importer.TLS.SkipVerify)
return importCmd

11
cmd/import_test.go
View file

@ -95,6 +95,17 @@ field = "f1"
return v.Error()
},
},
{
args: []string{"import", "--index", "i1", "--field", "f1", "--clear", "true"},
env: map[string]string{},
validation: func() error {
v := validator{}
v.Check(cmd.Importer.Index, "i1")
v.Check(cmd.Importer.Field, "f1")
v.Check(cmd.Importer.Clear, true)
return v.Error()
},
},
}
executeDry(t, tests)
}

9
ctl/import.go
View file

@ -49,6 +49,9 @@ type ImportCommand struct { // nolint: maligned
// CreateSchema ensures the schema exists before import
CreateSchema bool
// Clear clears the import data as opposed to setting it.
Clear bool
// Filenames to import from.
Paths []string `json:"paths"`
@ -255,7 +258,7 @@ func (cmd *ImportCommand) importBits(ctx context.Context, useColumnKeys, useRowK
// If keys are used, all bits are sent to the primary translate store (i.e. coordinator).
if useColumnKeys || useRowKeys {
logger.Printf("importing keys: n=%d", len(bits))
if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits); err != nil {
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
@ -272,7 +275,7 @@ func (cmd *ImportCommand) importBits(ctx context.Context, useColumnKeys, useRowK
}
logger.Printf("importing shard: %d, n=%d", shard, len(chunk))
if err := cmd.client.Import(ctx, cmd.Index, cmd.Field, shard, chunk); err != nil {
if err := cmd.client.Import(ctx, cmd.Index, cmd.Field, shard, chunk, pilosa.OptImportOptionsClear(cmd.Clear)); err != nil {
return errors.Wrap(err, "importing")
}
}
@ -377,7 +380,7 @@ func (cmd *ImportCommand) importValues(ctx context.Context, useColumnKeys bool,
}
logger.Printf("importing shard: %d, n=%d", shard, len(vals))
if err := cmd.client.ImportValue(ctx, cmd.Index, cmd.Field, shard, vals); err != nil {
if err := cmd.client.ImportValue(ctx, cmd.Index, cmd.Field, shard, vals, pilosa.OptImportOptionsClear(cmd.Clear)); err != nil {
return errors.Wrap(err, "importing values")
}
}

136
ctl/import_test.go
View file

@ -50,55 +50,111 @@ func TestImportCommand_Validation(t *testing.T) {
}
}
func TestImportCommand_Run(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import.csv")
file.Write([]byte("1,2\n3,4\n5,6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
func TestImportCommand_Basic(t *testing.T) {
t.Run("set", func(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import.csv")
file.Write([]byte("1,2\n3,4\n5,6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
cm.Index = "i"
cm.Field = "f"
cm.CreateSchema = true
cm.Paths = []string{file.Name()}
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run doesn't work: %s", err)
}
cm.Index = "i"
cm.Field = "f"
cm.CreateSchema = true
cm.Paths = []string{file.Name()}
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run doesn't work: %s", err)
}
})
t.Run("clear", func(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import.csv")
file.Write([]byte("1,2\n3,4\n5,6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
cm.Index = "i"
cm.Field = "f"
cm.CreateSchema = true
cm.Clear = true
cm.Paths = []string{file.Name()}
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run clear doesn't work: %s", err)
}
})
}
// Ensure that the ImportValue path runs.
func TestImportCommand_RunValue(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import-value.csv")
file.Write([]byte("1,2\n3,4\n5,6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
t.Run("set", func(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import-value.csv")
file.Write([]byte("1,2\n3,4\n5,6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader("")))
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"type": "int", "min": 0, "max": 100}}`)))
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader("")))
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"type": "int", "min": 0, "max": 100}}`)))
cm.Index = "i"
cm.Field = "f"
cm.Paths = []string{file.Name()}
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run with values doesn't work: %s", err)
}
cm.Index = "i"
cm.Field = "f"
cm.Paths = []string{file.Name()}
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run with values doesn't work: %s", err)
}
})
t.Run("clear", func(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import-value.csv")
file.Write([]byte("1,2\n3,4\n5,6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader("")))
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"type": "int", "min": 0, "max": 100}}`)))
cm.Index = "i"
cm.Field = "f"
cm.Paths = []string{file.Name()}
cm.Clear = true
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run with values doesn't work: %s", err)
}
})
}
// Ensure that import with keys runs.

12
docs/administration.md
View file

@ -82,6 +82,18 @@ For example, importing a file with the following contents will result in columns
<p>Note that you must first create a field. View <a href="../api-reference/#create-field">Create Field</a> for more details. The `-e` flag can create the necessary schema when using a field of type "set".</p>
</div>
#### Clearing Data via Import
By using the `--clear` flag with the import command, Pilosa will clear the values provided in the import payload.
For example, importing a file with the following contents along with the `--clear` flag will result in data being cleared from row 0, column 9; row 1, columns 2 and 8; and row 3, column 12. Clearing a value that doesn't exists is allowed.
```
0,9
1,2
1,8
3,12
```
#### Exporting
Exporting data to csv can be performed on a live instance of Pilosa. You need to specify the index and the field. The API also expects the shard number, but the `pilosa export` sub command will export all shards within a field. The data will be in csv format `Row,Column` and sorted by column.

174
docs/api-reference.md
View file

@ -11,14 +11,7 @@ nav = []
`GET /index`
Returns the schema of all indexes in JSON.
``` request
curl -XGET localhost:10101/index
```
``` response
{"indexes":[{"name":"user","fields":[{"name":"event","options":{"type":"time","timeQuantum":"YMD","keys":false}}]}]}
```
Is equivalent to `GET /schema` and returns the same response.
### List index schema
@ -30,7 +23,23 @@ Returns the schema of the specified index in JSON.
curl -XGET localhost:10101/index/user
```
``` response
{"name":"user","fields":[{"name":"event","options":{"type":"time","timeQuantum":"YMD","keys":false}}]}
{
"fields": [
{
"name": "event",
"options": {
"keys": false,
"timeQuantum": "YMD",
"type": "time"
}
}
],
"name": "user",
"options": {
"keys": false,
"trackExistence": true
}
}
```
### Create index
@ -39,8 +48,13 @@ curl -XGET localhost:10101/index/user
Creates an index with the given name.
The request payload is in JSON, and may contain the `options` field. The `options` field is a JSON object with the following options:
* `keys` (bool): Enables using column keys instead of column IDs.
* `trackExistence` (bool): Enables or disables existence tracking on the index. Required for [Not](../query-language/#not) queries. It is `true` by default.
``` request
curl -XPOST localhost:10101/index/user
curl -XPOST localhost:10101/index/user -d '{"options":{"keys":true}}'
```
``` response
{"success":true}
@ -71,7 +85,16 @@ curl localhost:10101/index/user/query \
-d 'Row(language=5)'
```
``` response
{"results":[{"attrs":{},"columns":[100]}]}
{
"results": [
{
"attrs": {},
"columns": [
100
]
}
]
}
```
In order to send protobuf binaries in the request and response, set `Content-Type` and `Accept` headers to: `application/x-protobuf`.
@ -87,20 +110,73 @@ curl "localhost:10101/index/user/query?columnAttrs=true&shards=0,1" \
```
``` response
{
"results":[{"attrs":{},"columns":[100]}],
"columnAttrs":[{"id":100,"attrs":{"name":"Klingon"}}]
"columnAttrs": [
{
"attrs": {
"name": "Klingon"
},
"id": 100
}
],
"results": [
{
"attrs": {},
"columns": [
100
]
}
]
}
```
By default, all bits and attributes (*for `Row` queries only*) are returned. In order to suppress returning bits, set `excludeBits` query argument to `true`; to suppress returning attributes, set `excludeAttrs` query argument to `true`.
### Import Data
`POST /index/<index-name>/field/<field-name>/import`
Supports high-rate data ingest to a particular shard of a particular field. The
official client libraries use this endpoint for their import functionality - it
is not usually necessary to use this endpoint directly. See the documentation for
imports for
<a href="https://github.com/pilosa/go-pilosa/blob/master/docs/imports-exports.md">Go</a>,
<a href="https://github.com/pilosa/java-pilosa/blob/master/docs/imports.md">Java</a>,
and <a href="https://github.com/pilosa/python-pilosa/tree/master/docs/imports.md">Python</a>.
The request payload is protobuf encoded with the following schema. The RowKeys
and/or ColumnKeys fields are used if the pilosa field or index are configured
for keys respectively. Otherwise, the RowIDs and ColumnIDs fields are used. They
must have the same number of items, and each index into those two lists
represents a particular bit to be set. Timestamps are optional, but if they
exist must also contain the same number of items as rows and columns. The
column IDs must all be in the shard specified in the request.
```
message ImportRequest {
string Index = 1;
string Field = 2;
uint64 Shard = 3;
repeated uint64 RowIDs = 4;
repeated uint64 ColumnIDs = 5;
repeated string RowKeys = 7;
repeated string ColumnKeys = 8;
repeated int64 Timestamps = 6;
}
```
### Create field
`POST /index/<index-name>/field/<field-name>`
Creates a field in the given index with the given name.
The request payload is in JSON, and may contain the `options` field. The `options` field is a JSON object which must contain a `type` along with the corresponding configuration options.
The request payload is in JSON, and may contain the `options` field. The `options` field is a JSON object which must contain a `type`:
* `type` (string): Sets the field type and type options.
* `keys` (bool): Enables using column keys instead of column IDs (optional).
Valid `type`s and correspondonding options are listed below:
* `set`
* `cacheType` (string): [ranked](../data-model/#ranked) or [LRU](../data-model/#lru) caching on this field. Default is `ranked`.
@ -158,6 +234,48 @@ curl -XDELETE localhost:10101/index/user/field/language
{"success":true}
```
### List all index schemas
`GET /schema`
Returns the schema of all indexes in JSON.
``` request
curl -XGET localhost:10101/index
```
``` response
{
"indexes": [
{
"fields": [
{
"name": "event",
"options": {
"keys": false,
"timeQuantum": "YMD",
"type": "time"
}
},
{
"name": "language",
"options": {
"cacheSize": 50000,
"cacheType": "ranked",
"keys": false,
"type": "set"
}
}
],
"name": "user",
"options": {
"keys": false,
"trackExistence": true
}
}
]
}
```
### Get version
`GET /version`
@ -171,6 +289,33 @@ curl -XGET localhost:10101/version
{"version":"v0.6.0"}
```
### Get status
`GET /status`
Returns the status of the cluster.
```request
curl -XGET localhost:10101/status
```
```response
{
"localID": "d3369125-29d8-4305-a351-b4474d14a542",
"nodes": [
{
"id": "d3369125-29d8-4305-a351-b4474d14a542",
"isCoordinator": true,
"uri": {
"host": "localhost",
"port": 10101,
"scheme": "http"
}
}
],
"state": "NORMAL"
}
```
### Recalculate Caches
`POST /recalculate-caches`
@ -187,4 +332,3 @@ curl -XPOST localhost:10101/recalculate-caches
```
Response: `204 No Content`

5
docs/getting-started.md
View file

@ -89,6 +89,11 @@ The `language` is a `set` field, but since the default field type is `set`, we d
#### Import Data From CSV Files
<div class="note">
<p>For demonstration purposes, we're using Pilosa's built in utility to import specially formatted CSV files. For more general usage, see how the various client libraries expose the bulk import functionality in <a href="https://github.com/pilosa/go-pilosa/blob/master/docs/imports-exports.md">Go</a>, <a href="https://github.com/pilosa/java-pilosa/blob/master/docs/imports.md">Java</a>, and <a href="https://github.com/pilosa/python-pilosa/tree/master/docs/imports.md">Python</a>. </p>
</div>
Download the `stargazer.csv` and `language.csv` files here:
```

4
docs/query-language.md
View file

@ -67,6 +67,10 @@ Set(<COLUMN>, <FIELD>=<ROW>, [TIMESTAMP])
`Set` assigns a value of 1 to a bit in the binary matrix, thus associating the given row (the `<ROW>` value) in the given field with the given column.
<div class="note">
<p>While using "Set" in PQL is a convenient way to get familiar with Pilosa, it's almost always better to use the import functionality in the <a href="https://github.com/pilosa/go-pilosa/blob/master/docs/imports-exports.md">Go</a>, <a href="https://github.com/pilosa/java-pilosa/blob/master/docs/imports.md">Java</a>, and <a href="https://github.com/pilosa/python-pilosa/tree/master/docs/imports.md">Python</a> clients to ingest lots of data. </p>
</div>
**Result Type:** boolean
A return value of `true` indicates that the bit was changed to 1.

76
encoding/proto/proto.go
View file

@ -384,6 +384,15 @@ func encodeQueryResponse(m *pilosa.QueryResponse) *internal.QueryResponse {
case bool:
pb.Results[i].Type = queryResultTypeBool
pb.Results[i].Changed = result
case pilosa.RowIDs:
pb.Results[i].Type = queryResultTypeRowIDs
pb.Results[i].RowIDs = result
case []pilosa.GroupCount:
pb.Results[i].Type = queryResultTypeGroupCounts
pb.Results[i].GroupCounts = encodeGroupCounts(result)
case pilosa.RowIdentifiers:
pb.Results[i].Type = queryResultTypeRowIdentifiers
pb.Results[i].RowIdentifiers = encodeRowIdentifiers(result)
case nil:
pb.Results[i].Type = queryResultTypeNil
}
@ -932,7 +941,6 @@ func decodeQueryResponse(pb *internal.QueryResponse, m *pilosa.QueryResponse) {
}
m.Results = make([]interface{}, len(pb.Results))
decodeQueryResults(pb.Results, m.Results)
}
func decodeColumnAttrSets(pb []*internal.ColumnAttrSet, m []*pilosa.ColumnAttrSet) {
@ -962,6 +970,9 @@ const (
queryResultTypeValCount
queryResultTypeUint64
queryResultTypeBool
queryResultTypeRowIDs
queryResultTypeGroupCounts
queryResultTypeRowIdentifiers
)
func decodeQueryResult(pb *internal.QueryResult) interface{} {
@ -978,6 +989,12 @@ func decodeQueryResult(pb *internal.QueryResult) interface{} {
return pb.Changed
case queryResultTypeNil:
return nil
case queryResultTypeRowIDs:
return pilosa.RowIDs(pb.RowIDs)
case queryResultTypeRowIdentifiers:
return decodeRowIdentifiers(pb.RowIdentifiers)
case queryResultTypeGroupCounts:
return decodeGroupCounts(pb.GroupCounts)
}
panic(fmt.Sprintf("unknown type: %d", pb.Type))
}
@ -1028,6 +1045,33 @@ func decodeAttr(attr *internal.Attr) (key string, value interface{}) {
}
}
func decodeRowIdentifiers(a *internal.RowIdentifiers) *pilosa.RowIdentifiers {
return &pilosa.RowIdentifiers{
Rows: a.Rows,
Keys: a.Keys,
}
}
func decodeGroupCounts(a []*internal.GroupCount) []pilosa.GroupCount {
other := make([]pilosa.GroupCount, len(a))
for i := range a {
other[i] = pilosa.GroupCount{
Group: decodeFieldRows(a[i].Group),
Count: a[i].Count,
}
}
return other
}
func decodeFieldRows(a []*internal.FieldRow) []pilosa.FieldRow {
other := make([]pilosa.FieldRow, len(a))
for i := range a {
other[i].Field = a[i].Field
other[i].RowID = a[i].RowID
}
return other
}
func decodePairs(a []*internal.Pair) []pilosa.Pair {
other := make([]pilosa.Pair, len(a))
for i := range a {
@ -1079,6 +1123,36 @@ func encodeRow(r *pilosa.Row) *internal.Row {
}
}
func encodeRowIdentifiers(r pilosa.RowIdentifiers) *internal.RowIdentifiers {
return &internal.RowIdentifiers{
Rows: r.Rows,
Keys: r.Keys,
//Attrs: encodeAttrs(r.Attrs),
}
}
func encodeGroupCounts(counts []pilosa.GroupCount) []*internal.GroupCount {
result := make([]*internal.GroupCount, len(counts))
for i := range counts {
result[i] = &internal.GroupCount{
Group: encodeFieldRows(counts[i].Group),
Count: counts[i].Count,
}
}
return result
}
func encodeFieldRows(a []pilosa.FieldRow) []*internal.FieldRow {
other := make([]*internal.FieldRow, len(a))
for i := range a {
other[i] = &internal.FieldRow{
Field: a[i].Field,
RowID: a[i].RowID,
}
}
return other
}
func encodePairs(a pilosa.Pairs) []*internal.Pair {
other := make([]*internal.Pair, len(a))
for i := range a {

579
executor.go
View file

@ -16,6 +16,7 @@ package pilosa
import (
"context"
"encoding/json"
"fmt"
"sort"
"time"
@ -257,6 +258,12 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
case "TopN":
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
return e.executeTopN(ctx, index, c, shards, opt)
case "Rows":
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
return e.executeRows(ctx, index, c, shards, opt)
case "GroupBy":
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
return e.executeGroupBy(ctx, index, c, shards, opt)
case "Options":
return e.executeOptionsCall(ctx, index, c, shards, opt)
default:
@ -777,14 +784,323 @@ func (e *executor) executeDifferenceShard(ctx context.Context, index string, c *
return other, nil
}
// RowIdentifiers is a return type for a list of
// row ids or row keys. The names `Rows` and `Keys`
// are meant to follow the same convention as the
// Row query which returns `Columns` and `Keys`.
// TODO: Rename this to something better. Anything.
type RowIdentifiers struct {
Rows []uint64 `json:"rows"`
Keys []string `json:"keys,omitempty"`
}
// RowIDs is a query return type for just uint64 row ids.
// It should only be used internally (since RowIdentifiers
// is the external return type), but it is exported because
// the proto package needs access to it.
type RowIDs []uint64
func (r RowIDs) merge(other RowIDs, limit int) RowIDs {
i, j := 0, 0
result := make(RowIDs, 0)
for i < len(r) && j < len(other) && len(result) < limit {
av, bv := r[i], other[j]
if av < bv {
result = append(result, av)
i++
} else if av > bv {
result = append(result, bv)
j++
} else {
result = append(result, bv)
i++
j++
}
}
for i < len(r) && len(result) < limit {
result = append(result, r[i])
i++
}
for j < len(other) && len(result) < limit {
result = append(result, other[j])
j++
}
return result
}
func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]GroupCount, error) {
// validate call
if len(c.Children) == 0 {
return nil, errors.New("need at least one child call")
}
limit := int(^uint(0) >> 1)
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, err
} else if hasLimit {
limit = int(lim)
}
// perform necessary Rows queries (any that have limit or columns args) -
// TODO, call async? would only help if multiple Rows queries had a column
// or limit arg.
// TODO support TopN in here would be really cool - and pretty easy I think.
childRows := make([]RowIDs, len(c.Children))
for i, child := range c.Children {
if child.Name != "Rows" {
return nil, errors.Errorf("'%s' is not a valid child query for GroupBy, must be 'Rows'", c.Name)
}
_, hasLimit, err := child.UintArg("limit")
if err != nil {
return nil, errors.Wrap(err, "getting limit")
}
_, hasCol, err := child.UintArg("column")
if err != nil {
return nil, errors.Wrap(err, "getting column")
}
if hasLimit || hasCol { // we need to perform this query cluster-wide ahead of executeGroupByShard
childRows[i], err = e.executeRows(ctx, index, child, shards, opt)
if err != nil {
return nil, errors.Wrap(err, "getting rows for ")
}
if len(childRows[i]) == 0 { // there are no results because this field has no values.
return []GroupCount{}, nil
}
}
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeGroupByShard(ctx, index, c, shard, childRows)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
other, _ := prev.([]GroupCount)
return mergeGroupCounts(other, v.([]GroupCount), limit)
}
// Get full result set.
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
if err != nil {
return nil, err
}
results, _ := other.([]GroupCount)
// Apply offset.
if offset, hasOffset, err := c.UintArg("offset"); err != nil {
return nil, err
} else if hasOffset {
if int(offset) < len(results) {
results = results[offset:]
}
}
// Apply limit.
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, err
} else if hasLimit {
if int(limit) < len(results) {
results = results[:limit]
}
}
return results, nil
}
// FieldRow is used to distinguish rows in a group by result.
type FieldRow struct {
Field string `json:"field"`
RowID uint64 `json:"rowID"`
RowKey string `json:"rowKey,omitempty"`
}
func (fr FieldRow) MarshalJSON() ([]byte, error) {
if fr.RowKey != "" {
return json.Marshal(struct {
Field string `json:"field"`
RowKey string `json:"rowKey"`
}{
Field: fr.Field,
RowKey: fr.RowKey,
})
}
return json.Marshal(struct {
Field string `json:"field"`
RowID uint64 `json:"rowID"`
}{
Field: fr.Field,
RowID: fr.RowID,
})
}
func (fr FieldRow) String() string {
return fmt.Sprintf("%s.%d", fr.Field, fr.RowID)
}
type GroupCount struct {
Group []FieldRow `json:"group"`
Count uint64 `json:"count"`
}
// mergeGroupCounts merges two slices of GroupCounts throwing away any that go
// beyond the limit. It assume that the two slices are sorted by the row ids in
// the fields of the group counts. It may modify its arguments.
func mergeGroupCounts(a, b []GroupCount, limit int) []GroupCount {
if limit > len(a)+len(b) {
limit = len(a) + len(b)
}
ret := make([]GroupCount, 0, limit)
i, j := 0, 0
for i < len(a) && j < len(b) && len(ret) < limit {
switch a[i].Compare(b[j]) {
case -1:
ret = append(ret, a[i])
i++
case 0:
a[i].Count += b[j].Count
ret = append(ret, a[i])
i++
j++
case 1:
ret = append(ret, b[j])
j++
}
}
for ; i < len(a) && len(ret) < limit; i++ {
ret = append(ret, a[i])
}
for ; j < len(b) && len(ret) < limit; j++ {
ret = append(ret, b[j])
}
return ret
}
func (g GroupCount) Compare(o GroupCount) int {
for i := range g.Group {
if g.Group[i].RowID < o.Group[i].RowID {
return -1
}
if g.Group[i].RowID > o.Group[i].RowID {
return 1
}
}
return 0
}
func (e *executor) executeGroupByShard(_ context.Context, index string, c *pql.Call, shard uint64, childRows []RowIDs) ([]GroupCount, error) {
iter, err := newGroupByIterator(childRows, c.Children, index, shard, e.Holder)
if err != nil {
return nil, errors.Wrapf(err, "getting group by iterator for shard %d", shard)
}
if iter == nil {
return []GroupCount{}, nil
}
limit := int(^uint(0) >> 1)
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, err
} else if hasLimit {
limit = int(lim)
}
results := make([]GroupCount, 0)
num := 0
for gc, done := iter.Next(); !done && num < limit; gc, done = iter.Next() {
if gc.Count > 0 {
num++
results = append(results, gc)
}
}
return results, nil
}
func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (RowIDs, error) {
if columnID, ok, err := c.UintArg("column"); err != nil {
return nil, errors.Wrap(err, "getting column")
} else if ok {
shards = []uint64{columnID / ShardWidth}
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeRowsShard(ctx, index, c, shard)
}
// Determine limit so we can use it when reducing.
limit := int(^uint(0) >> 1)
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, err
} else if hasLimit {
limit = int(lim)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
other, _ := prev.(RowIDs)
return other.merge(v.(RowIDs), limit)
}
// Get full result set.
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
if err != nil {
return nil, err
}
results, _ := other.(RowIDs)
return results, nil
}
func (e *executor) executeRowsShard(_ context.Context, index string, c *pql.Call, shard uint64) (RowIDs, error) {
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
return nil, ErrIndexNotFound
}
// Fetch field name from argument.
fieldName, ok := c.Args["field"].(string)
if !ok {
return nil, errors.New("Rows() argument required: field")
}
// Fetch field.
f := e.Holder.Field(index, fieldName)
if f == nil {
return nil, ErrFieldNotFound
}
frag := e.Holder.fragment(index, fieldName, viewStandard, shard)
if frag == nil {
return make(RowIDs, 0), nil
}
start := uint64(0)
if previous, ok, err := c.UintArg("previous"); err != nil {
return nil, errors.Wrap(err, "getting previous")
} else if ok {
start = previous + 1
}
filters := []rowFilter{}
if columnID, ok, err := c.UintArg("column"); err != nil {
return nil, err
} else if ok {
colShard := columnID >> shardWidthExponent
if colShard != shard {
return RowIDs{}, nil
}
filters = append(filters, filterColumn(columnID))
}
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, errors.Wrap(err, "getting limit")
} else if hasLimit {
filters = append(filters, filterWithLimit(limit))
}
return frag.rows(start, filters...), nil
}
func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
// Fetch column label from index.
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
return nil, ErrIndexNotFound
}
// Fetch field & row label based on argument.
// Fetch field name from argument.
fieldName, err := c.FieldArg()
if err != nil {
return nil, errors.New("Row() argument required: field")
@ -1867,6 +2183,12 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
// Positional args in new PQL syntax require special handling here.
rowKey = "_" + rowLabel
fieldName = callArgString(c, "_field")
case "Rows":
fieldName = callArgString(c, "field")
rowKey = "previous"
colKey = "column"
case "GroupBy":
return errors.Wrap(e.translateGroupByCall(index, idx, c), "translating GroupBy")
default:
colKey = "col"
fieldName = callArgString(c, "field")
@ -1942,6 +2264,61 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
return nil
}
func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call) error {
if c.Name != "GroupBy" {
panic("translateGroupByCall called with '" + c.Name + "'")
}
for _, child := range c.Children {
if err := e.translateCall(index, idx, child); err != nil {
return errors.Wrapf(err, "translating %s", child)
}
}
prev, ok := c.Args["previous"]
if !ok {
return nil // nothing else to be translated
}
previous, ok := prev.([]interface{})
if !ok {
return errors.Errorf("'previous' argument must be list, but got %T", prev)
}
if len(c.Children) != len(previous) {
return errors.Errorf("mismatched lengths for previous: %d and children: %d in %s", len(previous), len(c.Children), c)
}
fields := make([]*Field, len(c.Children))
for i, child := range c.Children {
fieldname := callArgString(child, "field")
field := idx.Field(fieldname)
if field == nil {
return errors.Wrapf(ErrFieldNotFound, "getting field '%s' from '%s'", fieldname, child)
}
fields[i] = field
}
for i, field := range fields {
prev := previous[i]
if field.keys() {
prevStr, ok := prev.(string)
if !ok {
return errors.New("prev value must be a string when field 'keys' option enabled")
}
ids, err := e.TranslateStore.TranslateRowsToUint64(index, field.Name(), []string{prevStr})
if err != nil {
return errors.Wrapf(err, "translating row key '%s'", prevStr)
}
previous[i] = ids[0]
} else {
if prevStr, ok := prev.(string); ok {
return errors.Errorf("got string row val '%s' in 'previous' for field %s which doesn't use string keys", prevStr, field.Name())
}
}
}
return nil
}
func (e *executor) translateResult(index string, idx *Index, call *pql.Call, result interface{}) (interface{}, error) {
switch result := result.(type) {
case *Row:
@ -1977,7 +2354,62 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res
return other, nil
}
}
case []GroupCount:
other := make([]GroupCount, 0)
for _, gl := range result {
group := make([]FieldRow, len(gl.Group))
for i, g := range gl.Group {
group[i] = g
// TODO: It may be useful to cache this field lookup.
field := idx.Field(g.Field)
if field == nil {
return nil, ErrFieldNotFound
}
if field.keys() {
key, err := e.TranslateStore.TranslateRowToString(index, g.Field, g.RowID)
if err != nil {
return nil, errors.Wrap(err, "translating row ID in Group")
}
group[i].RowKey = key
}
}
other = append(other, GroupCount{
Group: group,
Count: gl.Count,
})
}
return other, nil
case RowIDs:
other := RowIdentifiers{}
fieldName := callArgString(call, "field")
if fieldName == "" {
return nil, ErrFieldNotFound
}
if field := idx.Field(fieldName); field == nil {
return nil, ErrFieldNotFound
} else if field.keys() {
other.Keys = make([]string, len(result))
for i, id := range result {
key, err := e.TranslateStore.TranslateRowToString(index, fieldName, id)
if err != nil {
return nil, errors.Wrap(err, "translating row ID")
}
other.Keys[i] = key
}
} else {
other.Rows = result
}
return other, nil
}
return result, nil
}
@ -2026,7 +2458,7 @@ func needsShards(calls []*pql.Call) bool {
switch call.Name {
case "Clear", "Set", "SetRowAttrs", "SetColumnAttrs":
continue
case "Count", "TopN":
case "Count", "TopN", "Rows":
return true
// default catches Bitmap calls
default:
@ -2096,3 +2528,144 @@ func isString(v interface{}) bool {
_, ok := v.(string)
return ok
}
// groupByIterator contains several slices. Each slice contains a number of
// elements equal to the number of fields in the group by (the number of Rows
// calls).
type groupByIterator struct {
// rowIters contains a rowIterator for each of the fields in the Group By.
rowIters []*rowIterator
// rows contains the current row data for each of the fields in the Group
// By. Each row is the intersection of itself and the rows of the fields
// with an index lower than its own. This is a performance optimization so
// that the expected common case of getting the next row in the furthest
// field to the right require only a single intersect with the row of the
// previous field to determine the count of the new group.
rows []struct {
row *Row
id uint64
}
// fields helps with the construction of GroupCount results by holding all
// the field names that are being grouped by. Each results makes a copy of
// fields and then sets the row ids.
fields []FieldRow
done bool
}
// newGroupByIterator initializes a new groupByIterator.
func newGroupByIterator(rowIDs []RowIDs, children []*pql.Call, index string, shard uint64, holder *Holder) (*groupByIterator, error) {
gbi := &groupByIterator{
rowIters: make([]*rowIterator, len(children)),
rows: make([]struct {
row *Row
id uint64
}, len(children)),
fields: make([]FieldRow, len(children)),
}
ignorePrev := false
for i, call := range children {
fieldName := call.Args["field"].(string) // this has already been validated by this point
gbi.fields[i].Field = fieldName
// Fetch fragment.
frag := holder.fragment(index, fieldName, viewStandard, shard)
if frag == nil { // this means this whole shard doesn't have all it needs to continue
return nil, nil
}
filters := []rowFilter{}
if len(rowIDs[i]) > 0 {
filters = append(filters, filterWithRows(rowIDs[i]))
}
gbi.rowIters[i] = frag.rowIterator(i != 0, filters...)
prev, hasPrev, err := call.UintArg("previous")
if err != nil {
return nil, errors.Wrap(err, "getting previous")
} else if hasPrev && !ignorePrev {
if i == len(children)-1 {
prev += 1
}
gbi.rowIters[i].Seek(prev)
}
nextRow, rowID, wrapped := gbi.rowIters[i].Next()
if nextRow == nil {
gbi.done = true
return gbi, nil
}
gbi.rows[i].row = nextRow
gbi.rows[i].id = rowID
if hasPrev && rowID != prev {
// ignorePrev signals that we didn't find a previous row, so all
// Rows queries "deeper" than it need to ignore the previous
// argument and start at the beginning.
ignorePrev = true
}
if wrapped {
// if a field has wrapped, we need to get the next row for the
// previous field, and if that one wraps we need to keep going
// backward.
for j := i - 1; j >= 0; j-- {
nextRow, rowID, wrapped := gbi.rowIters[j].Next()
if nextRow == nil {
gbi.done = true
return gbi, nil
}
gbi.rows[j].row = nextRow
gbi.rows[j].id = rowID
if !wrapped {
break
}
}
}
}
for i := 1; i < len(gbi.rows)-1; i++ {
gbi.rows[i].row = gbi.rows[i].row.Intersect(gbi.rows[i-1].row)
}
return gbi, nil
}
// nextAtIdx is a recursive helper method for getting the next row for the field
// at index i, and then updating the rows in the "higher" fields if it wraps.
func (gbi *groupByIterator) nextAtIdx(i int) {
nr, rowID, wrapped := gbi.rowIters[i].Next()
if nr == nil {
gbi.done = true
return
}
if wrapped && i != 0 {
gbi.nextAtIdx(i - 1)
}
if i != 0 && i != len(gbi.rows)-1 {
gbi.rows[i].row = nr.Intersect(gbi.rows[i-1].row)
} else {
gbi.rows[i].row = nr
}
gbi.rows[i].id = rowID
}
// Next returns a GroupCount representing the next group by record. When there
// are no more records it will return an empty GroupCount and done==true.
func (gbi *groupByIterator) Next() (ret GroupCount, done bool) {
if gbi.done {
return ret, true
}
if len(gbi.rows) == 1 {
ret.Count = gbi.rows[len(gbi.rows)-1].row.Count()
} else {
ret.Count = gbi.rows[len(gbi.rows)-1].row.intersectionCount(gbi.rows[len(gbi.rows)-2].row)
}
ret.Group = make([]FieldRow, len(gbi.rows))
copy(ret.Group, gbi.fields)
for i, r := range gbi.rows {
ret.Group[i].RowID = r.id
}
// set up for next call
gbi.nextAtIdx(len(gbi.rows) - 1)
return ret, false
}

224
executor_internal_test.go Normal file
View file

@ -0,0 +1,224 @@
package pilosa
import (
"encoding/json"
"fmt"
"io/ioutil"
"strings"
"testing"
"github.com/pilosa/pilosa/pql"
)
func TestExecutor_TranslateGroupByCall(t *testing.T) {
e := &executor{
Holder: NewHolder(),
}
e.Holder.Path, _ = ioutil.TempDir("", "")
err := e.Holder.Open()
if err != nil {
t.Fatalf("opening holder: %v", err)
}
e.TranslateStore = e.Holder.translateFile
tf, _ := ioutil.TempFile("", "")
e.Holder.translateFile.Path = tf.Name()
err = e.Holder.translateFile.Open()
if err != nil {
t.Fatalf("opening translateFile: %v", err)
}
idx, err := e.Holder.CreateIndex("i", IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, erra := idx.CreateField("ak", OptFieldKeys())
_, errb := idx.CreateField("b")
_, errc := idx.CreateField("ck", OptFieldKeys())
if erra != nil || errb != nil || errc != nil {
t.Fatalf("creating fields %v, %v, %v", erra, errb, errc)
}
_, erra = e.TranslateStore.TranslateRowsToUint64("i", "ak", []string{"la"})
_, errb = e.TranslateStore.TranslateRowsToUint64("i", "ck", []string{"ha"})
if erra != nil || errb != nil {
t.Fatalf("translating rows %v, %v", erra, errb)
}
query, err := pql.ParseString(`GroupBy(Rows(field=ak), Rows(field=b), Rows(field=ck), previous=["la", 0, "ha"])`)
if err != nil {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
err = e.translateGroupByCall("i", idx, c)
if err != nil {
t.Fatalf("translating call: %v", err)
}
if len(c.Args["previous"].([]interface{})) != 3 {
t.Fatalf("unexpected length for 'previous' arg %v", c.Args["previous"])
}
for i, v := range c.Args["previous"].([]interface{}) {
if !isInt(v) {
t.Fatalf("expected all items in previous to be ints, but '%v' at index %d is %[1]T", v, i)
}
}
errTests := []struct {
pql string
err string
}{
{
pql: `GroupBy(Rows(field=notfound), previous=1)`,
err: "'previous' argument must be list",
},
{
pql: `GroupBy(Rows(field=ak), previous=["la", 0])`,
err: "mismatched lengths",
},
{
pql: `GroupBy(Rows(field=ak), previous=[1])`,
err: "prev value must be a string",
},
{
pql: `GroupBy(Rows(field=notfound), previous=[1])`,
err: ErrFieldNotFound.Error(),
},
// TODO: an unknown key will actually allocate an id. this is probably bad.
// {
// pql: `GroupBy(Rows(field=ak), previous=["zoop"])`,
// err: "translating row key '",
// },
{
pql: `GroupBy(Rows(field=b), previous=["la"])`,
err: "which doesn't use string keys",
},
}
for i, test := range errTests {
t.Run(fmt.Sprintf("#%d_%s", i, test.err), func(t *testing.T) {
query, err := pql.ParseString(test.pql)
if err != nil {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
err = e.translateGroupByCall("i", idx, c)
if err == nil {
t.Fatalf("expected error, but translated call is '%s", c)
}
if !strings.Contains(err.Error(), test.err) {
t.Fatalf("expected '%s', got '%v'", test.err, err)
}
})
}
}
func isInt(a interface{}) bool {
switch a.(type) {
case int, int64, uint, uint64:
return true
default:
return false
}
}
func TestFilterWithLimit(t *testing.T) {
f := filterWithLimit(5)
for i := uint64(0); i < 5; i++ {
include, done := f(i, i*(1<<shardVsContainerExponent), nil)
if done {
t.Fatalf("limit filter ended early on iteration %d", i)
}
if !include {
t.Fatalf("limit filter should always include until done")
}
}
inc, done := f(5, 5*(1<<shardVsContainerExponent)+1, nil)
if !done {
t.Fatalf("limit filter should have been done, but got inc: %v done: %v", inc, done)
}
}
func TestFilterWithRows(t *testing.T) {
tests := []struct {
rows []uint64
callWith []uint64
expect [][2]bool
}{
{
rows: []uint64{},
callWith: []uint64{0},
expect: [][2]bool{{false, true}},
},
{
rows: []uint64{0},
callWith: []uint64{0},
expect: [][2]bool{{true, true}},
},
{
rows: []uint64{1},
callWith: []uint64{0, 2},
expect: [][2]bool{{false, false}, {false, true}},
},
{
rows: []uint64{0},
callWith: []uint64{1, 2},
expect: [][2]bool{{false, true}, {false, true}},
},
{
rows: []uint64{3, 9},
callWith: []uint64{1, 2, 3, 10},
expect: [][2]bool{{false, false}, {false, false}, {true, false}, {false, true}},
},
{
rows: []uint64{0, 1, 2},
callWith: []uint64{0, 1, 2},
expect: [][2]bool{{true, false}, {true, false}, {true, true}},
},
}
for num, test := range tests {
t.Run(fmt.Sprintf("%d_%v_with_%v", num, test.rows, test.callWith), func(t *testing.T) {
if len(test.callWith) != len(test.expect) {
t.Fatalf("Badly specified test - must expect the same number of values as calls.")
}
f := filterWithRows(test.rows)
for i, id := range test.callWith {
inc, done := f(id, 0, nil)
if inc != test.expect[i][0] || done != test.expect[i][1] {
t.Fatalf("Calling with %d\nexp: %v,%v\ngot: %v,%v", id, test.expect[i][0], test.expect[i][1], inc, done)
}
}
})
}
}
func TestFieldRowMarshalJSON(t *testing.T) {
fr := FieldRow{
Field: "blah",
RowID: 0,
RowKey: "ha",
}
b, err := json.Marshal(fr)
if err != nil {
t.Fatalf("marshalling fieldrow: %v", err)
}
if string(b) != `{"field":"blah","rowKey":"ha"}` {
t.Fatalf("unexpected json: %s", b)
}
fr = FieldRow{
Field: "blah",
RowID: 2,
RowKey: "",
}
b, err = json.Marshal(fr)
if err != nil {
t.Fatalf("marshalling fieldrow: %v", err)
}
if string(b) != `{"field":"blah","rowID":2}` {
t.Fatalf("unexpected json: %s", b)
}
}

502
executor_test.go
View file

@ -1894,6 +1894,22 @@ Set(4500001, fn=4)
t.Fatalf("wrong attrs: %v", attrst)
}
})
t.Run("remote groupBy", func(t *testing.T) {
if res, err := c[1].API.Query(context.Background(), &pilosa.QueryRequest{
Index: "i",
Query: `GroupBy(Rows(field=f))`,
}); err != nil {
t.Fatalf("GroupBy querying: %v", err)
} else {
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "f", RowID: 7}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "f", RowID: 10}}, Count: 4},
}
results := res.Results[0].([]pilosa.GroupCount)
checkGroupBy(t, expected, results)
}
})
}
// Ensure executor returns an error if too many writes are in a single request.
@ -2623,6 +2639,492 @@ func benchmarkExistence(nn bool, b *testing.B) {
func BenchmarkExecutor_Existence_True(b *testing.B) { benchmarkExistence(true, b) }
func BenchmarkExecutor_Existence_False(b *testing.B) { benchmarkExistence(false, b) }
func TestExecutor_Execute_Rows(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
c.CreateField(t, "i", pilosa.IndexOptions{}, "general")
c.ImportBits(t, "i", "general", [][2]uint64{
{10, 0},
{10, ShardWidth + 1},
{11, 2},
{11, ShardWidth + 2},
{12, 2},
{12, ShardWidth + 2},
{13, 3},
})
rows := c.Query(t, "i", `Rows(field=general)`).Results[0].(pilosa.RowIdentifiers)
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{10, 11, 12, 13}}) {
t.Fatalf("unexpected rows: %+v", rows)
}
rows = c.Query(t, "i", `Rows(field=general, limit=2)`).Results[0].(pilosa.RowIdentifiers)
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{10, 11}}) {
t.Fatalf("unexpected rows: %+v", rows)
}
rows = c.Query(t, "i", `Rows(field=general, previous=10,limit=2)`).Results[0].(pilosa.RowIdentifiers)
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{11, 12}}) {
t.Fatalf("unexpected rows: %+v", rows)
}
rows = c.Query(t, "i", `Rows(field=general, column=2)`).Results[0].(pilosa.RowIdentifiers)
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{11, 12}}) {
t.Fatalf("unexpected rows: %+v", rows)
}
}
func TestExecutor_Execute_Rows_Keys(t *testing.T) {
c := test.MustRunCluster(t, 1)
defer c.Close()
_, err := c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{Keys: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldKeys())
if err != nil {
t.Fatalf("creating field: %v", err)
}
// setup some data. 10 bits in each of shards 0 through 9. starting at
// row/col shardNum and progressing to row/col shardNum+10. Also set the
// previous 2 for each bit if row >0.
query := strings.Builder{}
for shard := 0; shard < 10; shard++ {
for i := shard; i < shard+10; i++ {
for row := i; row >= 0 && row > i-3; row-- {
query.WriteString(fmt.Sprintf("Set(\"%d\", f=\"%d\")", shard*pilosa.ShardWidth+i, row))
}
}
}
_, err = c[0].API.Query(context.Background(), &pilosa.QueryRequest{
Index: "i",
Query: query.String(),
})
if err != nil {
t.Fatalf("querying: %v", err)
}
tests := []struct {
q string
exp []string
}{
{
q: `Rows(field=f)`,
exp: []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18"},
},
{
q: `Rows(field=f, limit=2)`,
exp: []string{"0", "1"},
},
{
q: `Rows(field=f, previous="15")`,
exp: []string{"16", "17", "18"},
},
{
q: `Rows(field=f, previous="11", limit=2)`,
exp: []string{"12", "13"},
},
{
q: `Rows(field=f, previous="17", limit=5)`,
exp: []string{"18"},
},
{
q: `Rows(field=f, previous="18")`,
exp: []string{},
},
{
q: `Rows(field=f, previous="1", limit=0)`,
exp: []string{},
},
{
q: `Rows(field=f, column="1")`,
exp: []string{"0", "1"},
},
{
q: `Rows(field=f, column="2")`,
exp: []string{"0", "1", "2"},
},
{
q: `Rows(field=f, column="3")`,
exp: []string{"1", "2", "3"},
},
{
q: `Rows(field=f, limit=2, column="3")`,
exp: []string{"1", "2"},
},
{
q: fmt.Sprintf(`Rows(field=f, previous="15", column="%d")`, ShardWidth*9+17),
exp: []string{"16", "17"},
},
{
q: fmt.Sprintf(`Rows(field=f, previous="11", limit=2, column="%d")`, ShardWidth*5+14),
exp: []string{"12", "13"},
},
{
q: fmt.Sprintf(`Rows(field=f, previous="17", limit=5, column="%d")`, ShardWidth*9+18),
exp: []string{"18"},
},
{
q: `Rows(field=f, previous="18", column="19")`,
exp: []string{},
},
{
q: `Rows(field=f, previous="1", limit=0, column="0")`,
exp: []string{},
},
}
for i, test := range tests {
t.Run(fmt.Sprintf("#%d_%s", i, test.q), func(t *testing.T) {
if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: test.q}); err != nil {
t.Fatal(err)
} else if rows := res.Results[0].(pilosa.RowIdentifiers); !reflect.DeepEqual(
rows, pilosa.RowIdentifiers{Keys: test.exp}) {
t.Fatalf("\ngot: %+v\nexp: %+v", rows, pilosa.RowIdentifiers{Keys: test.exp})
}
})
}
}
func TestExecutor_Execute_GroupBy(t *testing.T) {
groupByTest := func(t *testing.T, clusterSize int) {
c := test.MustRunCluster(t, 1)
defer c.Close()
c.CreateField(t, "i", pilosa.IndexOptions{}, "general")
c.CreateField(t, "i", pilosa.IndexOptions{}, "sub")
c.ImportBits(t, "i", "general", [][2]uint64{
{10, 0},
{10, 1},
{10, ShardWidth + 1},
{11, 2},
{11, ShardWidth + 2},
{12, 2},
{12, ShardWidth + 2},
})
c.ImportBits(t, "i", "sub", [][2]uint64{
{100, 0},
{100, 1},
{100, 3},
{100, ShardWidth + 1},
{110, 2},
{110, 0},
})
t.Run("No Field List Arguments", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy()`}); err != nil {
if !strings.Contains(err.Error(), "need at least one child call") {
t.Fatalf("unexpected error: \"%v\"", err)
}
}
})
t.Run("Unknown Field ", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(Rows(field=missing))`}); err != nil {
if errors.Cause(err) != pilosa.ErrFieldNotFound {
t.Fatalf("unexpected error\n\"%s\" not returned instead \n\"%s\"", pilosa.ErrFieldNotFound, err)
}
}
})
t.Run("Basic", func(t *testing.T) {
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3},
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}, {Field: "sub", RowID: 110}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1},
}
results := c.Query(t, "i", `GroupBy(Rows(field=general), Rows(field=sub))`).Results[0].([]pilosa.GroupCount)
checkGroupBy(t, expected, results)
})
t.Run("check field offset no limit", func(t *testing.T) {
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}}, Count: 2},
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}}, Count: 2},
}
results := c.Query(t, "i", `GroupBy(Rows(field=general, previous=10))`).Results[0].([]pilosa.GroupCount)
checkGroupBy(t, expected, results)
})
t.Run("check field offset limit", func(t *testing.T) {
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}}, Count: 2},
}
results := c.Query(t, "i", `GroupBy(Rows(field=general, previous=10), limit=1)`).Results[0].([]pilosa.GroupCount)
checkGroupBy(t, expected, results)
})
c.CreateField(t, "i", pilosa.IndexOptions{}, "a")
c.CreateField(t, "i", pilosa.IndexOptions{}, "b")
c.ImportBits(t, "i", "a", [][2]uint64{
{0, 1},
{1, ShardWidth + 1},
})
c.ImportBits(t, "i", "b", [][2]uint64{
{0, ShardWidth + 1},
{1, 1},
})
t.Run("tricky data", func(t *testing.T) {
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "a", RowID: 0}, {Field: "b", RowID: 1}}, Count: 1},
}
results := c.Query(t, "i", `GroupBy(Rows(field=a), Rows(field=b), limit=1)`).Results[0].([]pilosa.GroupCount)
checkGroupBy(t, expected, results)
})
// set the same bits in a single shard in three fields
c.CreateField(t, "i", pilosa.IndexOptions{}, "wa")
c.CreateField(t, "i", pilosa.IndexOptions{}, "wb")
c.CreateField(t, "i", pilosa.IndexOptions{}, "wc")
c.ImportBits(t, "i", "wa", [][2]uint64{
{0, 0}, {0, 1}, {0, 2}, // all
{1, 1}, // odds
{2, 0}, {2, 2}, // evens
{3, 3}, // no overlap
})
c.ImportBits(t, "i", "wb", [][2]uint64{
{0, 0}, {0, 1}, {0, 2},
{1, 1},
{2, 0}, {2, 2},
{3, 3},
})
c.ImportBits(t, "i", "wc", [][2]uint64{
{0, 0}, {0, 1}, {0, 2},
{1, 1},
{2, 0}, {2, 2},
{3, 3},
})
t.Run("test wrapping with previous", func(t *testing.T) {
results := c.Query(t, "i", `GroupBy(Rows(field=wa), Rows(field=wb), Rows(field=wc, previous=1), limit=3)`).Results[0].([]pilosa.GroupCount)
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 0}, {Field: "wb", RowID: 0}, {Field: "wc", RowID: 2}}, Count: 2},
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 0}, {Field: "wb", RowID: 1}, {Field: "wc", RowID: 0}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 0}, {Field: "wb", RowID: 1}, {Field: "wc", RowID: 1}}, Count: 1},
}
checkGroupBy(t, expected, results)
})
t.Run("test previous is last result", func(t *testing.T) {
results := c.Query(t, "i", `GroupBy(Rows(field=wa, previous=3), Rows(field=wb, previous=3), Rows(field=wc, previous=3), limit=3)`).Results[0].([]pilosa.GroupCount)
if len(results) > 0 {
t.Fatalf("expected no results because previous specified last result")
}
})
t.Run("test wrapping multiple", func(t *testing.T) {
results := c.Query(t, "i", `GroupBy(Rows(field=wa), Rows(field=wb, previous=2), Rows(field=wc, previous=2), limit=1)`).Results[0].([]pilosa.GroupCount)
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 1}, {Field: "wb", RowID: 0}, {Field: "wc", RowID: 0}}, Count: 1},
}
checkGroupBy(t, expected, results)
})
// test multiple shards with distinct results (different rows) and same
// rows to ensure ordering, limit behavior and correctness
c.CreateField(t, "i", pilosa.IndexOptions{}, "ma")
c.CreateField(t, "i", pilosa.IndexOptions{}, "mb")
c.ImportBits(t, "i", "ma", [][2]uint64{
{0, 0},
{1, ShardWidth},
{2, 0},
{3, ShardWidth},
})
c.ImportBits(t, "i", "mb", [][2]uint64{
{0, 0},
{1, ShardWidth},
{2, 0},
{3, ShardWidth},
})
t.Run("distinct rows in different shards", func(t *testing.T) {
results := c.Query(t, "i", `GroupBy(Rows(field=ma), Rows(field=mb), limit=5)`).Results[0].([]pilosa.GroupCount)
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 0}, {Field: "mb", RowID: 0}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 0}, {Field: "mb", RowID: 2}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 1}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 3}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 2}, {Field: "mb", RowID: 0}}, Count: 1},
}
checkGroupBy(t, expected, results)
})
t.Run("distinct rows in different shards with row limit", func(t *testing.T) {
results := c.Query(t, "i", `GroupBy(Rows(field=ma), Rows(field=mb, limit=2), limit=5)`).Results[0].([]pilosa.GroupCount)
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 0}, {Field: "mb", RowID: 0}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 1}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 2}, {Field: "mb", RowID: 0}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 3}, {Field: "mb", RowID: 1}}, Count: 1},
}
checkGroupBy(t, expected, results)
})
t.Run("distinct rows in different shards with column arg", func(t *testing.T) {
results := c.Query(t, "i", fmt.Sprintf(`GroupBy(Rows(field=ma), Rows(field=mb, column=%d), limit=5)`, ShardWidth)).Results[0].([]pilosa.GroupCount)
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 1}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 3}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 3}, {Field: "mb", RowID: 1}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 3}, {Field: "mb", RowID: 3}}, Count: 1},
}
checkGroupBy(t, expected, results)
})
c.CreateField(t, "i", pilosa.IndexOptions{}, "na")
c.CreateField(t, "i", pilosa.IndexOptions{}, "nb")
c.ImportBits(t, "i", "na", [][2]uint64{
{0, 0},
{0, ShardWidth},
{1, 0},
{1, ShardWidth},
})
c.ImportBits(t, "i", "nb", [][2]uint64{
{0, 0},
{0, ShardWidth},
{1, 0},
{1, ShardWidth},
})
t.Run("same rows in different shards", func(t *testing.T) {
results := c.Query(t, "i", `GroupBy(Rows(field=na), Rows(field=nb))`).Results[0].([]pilosa.GroupCount)
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "na", RowID: 0}, {Field: "nb", RowID: 0}}, Count: 2},
{Group: []pilosa.FieldRow{{Field: "na", RowID: 0}, {Field: "nb", RowID: 1}}, Count: 2},
{Group: []pilosa.FieldRow{{Field: "na", RowID: 1}, {Field: "nb", RowID: 0}}, Count: 2},
{Group: []pilosa.FieldRow{{Field: "na", RowID: 1}, {Field: "nb", RowID: 1}}, Count: 2},
}
checkGroupBy(t, expected, results)
})
// test paging over results using previous. set the same bits in three
// fields
c.CreateField(t, "i", pilosa.IndexOptions{}, "ppa")
c.CreateField(t, "i", pilosa.IndexOptions{}, "ppb")
c.CreateField(t, "i", pilosa.IndexOptions{}, "ppc")
c.ImportBits(t, "i", "ppa", [][2]uint64{
{0, 0},
{1, 0},
{2, 0},
{3, 0}, {3, 91000}, {3, ShardWidth}, {3, ShardWidth * 2}, {3, ShardWidth * 3},
})
c.ImportBits(t, "i", "ppb", [][2]uint64{
{0, 0},
{1, 0},
{2, 0},
{3, 0}, {3, 91000}, {3, ShardWidth}, {3, ShardWidth * 2}, {3, ShardWidth * 3},
})
c.ImportBits(t, "i", "ppc", [][2]uint64{
{0, 0},
{1, 0},
{2, 0},
{3, 0}, {3, 91000}, {3, ShardWidth}, {3, ShardWidth * 2}, {3, ShardWidth * 3},
})
t.Run("test wrapping with previous", func(t *testing.T) {
totalResults := make([]pilosa.GroupCount, 0)
results := c.Query(t, "i", `GroupBy(Rows(field=ppa), Rows(field=ppb), Rows(field=ppc), limit=3)`).Results[0].([]pilosa.GroupCount)
totalResults = append(totalResults, results...)
for len(totalResults) < 64 {
lastGroup := results[len(results)-1].Group
query := fmt.Sprintf("GroupBy(Rows(field=ppa, previous=%d), Rows(field=ppb, previous=%d), Rows(field=ppc, previous=%d), limit=3)", lastGroup[0].RowID, lastGroup[1].RowID, lastGroup[2].RowID)
results = c.Query(t, "i", query).Results[0].([]pilosa.GroupCount)
totalResults = append(totalResults, results...)
}
expected := make([]pilosa.GroupCount, 64)
for i := 0; i < 64; i++ {
expected[i] = pilosa.GroupCount{Group: []pilosa.FieldRow{{Field: "ppa", RowID: uint64(i / 16)}, {Field: "ppb", RowID: uint64((i % 16) / 4)}, {Field: "ppc", RowID: uint64(i % 4)}}, Count: 1}
}
expected[63].Count = 5
checkGroupBy(t, expected, totalResults)
})
}
for size := range []int{1, 3} {
t.Run(fmt.Sprintf("%d_nodes", size), func(t *testing.T) {
groupByTest(t, size)
})
}
}
func BenchmarkGroupBy(b *testing.B) {
c := test.MustRunCluster(b, 1)
defer c.Close()
c.CreateField(b, "i", pilosa.IndexOptions{}, "a")
c.CreateField(b, "i", pilosa.IndexOptions{}, "b")
c.CreateField(b, "i", pilosa.IndexOptions{}, "c")
// Set up identical representative data in 3 fields. In each row, we'll set
// a certain bit pattern for 100 bits, then skip 1000 up to ShardWidth.
bits := make([][2]uint64, 0)
for i := uint64(0); i < ShardWidth; i++ {
// row 0 has 100 bit runs
bits = append(bits, [2]uint64{0, i})
if i%2 == 1 {
// row 1 has odd bits set
bits = append(bits, [2]uint64{1, i})
}
if i%2 == 0 {
// row 2 has even bits set
bits = append(bits, [2]uint64{2, i})
}
if i%27 == 0 {
// row 3 has every 27th bit set
bits = append(bits, [2]uint64{3, i})
}
if i%100 == 99 {
i += 1000
}
}
c.ImportBits(b, "i", "a", bits)
c.ImportBits(b, "i", "b", bits)
c.ImportBits(b, "i", "c", bits)
b.Run("single shard group by", func(b *testing.B) {
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
c.Query(b, "i", `GroupBy(Rows(field=a), Rows(field=b), Rows(field=c))`)
}
})
b.Run("single shard with limit", func(b *testing.B) {
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
c.Query(b, "i", `GroupBy(Rows(field=a), Rows(field=b), Rows(field=c), limit=4)`)
}
})
// TODO benchmark over multiple shards
// TODO benchmark paging over large numbers of rows
}
func checkGroupBy(t *testing.T, expected, results []pilosa.GroupCount) {
if len(results) != len(expected) {
t.Fatalf("number of groupings mismatch:\n got:%+v\nwant:%+v\n", results, expected)
}
for i, result := range results {
if !reflect.DeepEqual(expected[i], result) {
t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
}
}
}
func runCallTest(t *testing.T, writeQuery string, readQueries []string, indexOptions *pilosa.IndexOptions, fieldOption ...pilosa.FieldOption) []pilosa.QueryResponse {
if indexOptions == nil {
indexOptions = &pilosa.IndexOptions{}

30
field.go
View file

@ -1047,11 +1047,25 @@ func (f *Field) Range(name string, op pql.Token, predicate int64) (*Row, error)
}
// Import bulk imports data.
func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) error {
func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time, opts ...ImportOption) error {
// Set up import options.
options := &ImportOptions{}
for _, opt := range opts {
err := opt(options)
if err != nil {
return errors.Wrap(err, "applying option")
}
}
// Determine quantum if timestamps are set.
q := f.TimeQuantum()
if hasTime(timestamps) && q == "" {
return errors.New("time quantum not set in field")
if hasTime(timestamps) {
if q == "" {
return errors.New("time quantum not set in field")
} else if options.Clear {
return errors.New("import clear is not supported with timestamps")
}
}
fieldType := f.Type()
@ -1103,7 +1117,7 @@ func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) erro
return errors.Wrap(err, "creating view")
}
if err := frag.bulkImport(data.RowIDs, data.ColumnIDs); err != nil {
if err := frag.bulkImport(data.RowIDs, data.ColumnIDs, options); err != nil {
return err
}
}
@ -1112,7 +1126,7 @@ func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) erro
}
// importValue bulk imports range-encoded value data.
func (f *Field) importValue(columnIDs []uint64, values []int64) error {
func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportOptions) error {
viewName := viewBSIGroupPrefix + f.name
// Get the bsiGroup so we know bitDepth.
bsig := f.bsiGroup(f.name)
@ -1160,7 +1174,7 @@ func (f *Field) importValue(columnIDs []uint64, values []int64) error {
baseValues[i] = uint64(value - bsig.Min)
}
if err := frag.importValue(data.ColumnIDs, baseValues, bsig.BitDepth()); err != nil {
if err := frag.importValue(data.ColumnIDs, baseValues, bsig.BitDepth(), options.Clear); err != nil {
return err
}
}
@ -1168,7 +1182,7 @@ func (f *Field) importValue(columnIDs []uint64, values []int64) error {
return nil
}
func (f *Field) importRoaring(data []byte, shard uint64) error {
func (f *Field) importRoaring(data []byte, shard uint64, clear bool) error {
viewName := viewStandard
view, err := f.createViewIfNotExists(viewName)
@ -1181,7 +1195,7 @@ func (f *Field) importRoaring(data []byte, shard uint64) error {
return errors.Wrap(err, "creating fragment")
}
if err := frag.importRoaring(data); err != nil {
if err := frag.importRoaring(data, clear); err != nil {
return err
}

267
fragment.go
View file

@ -58,6 +58,9 @@ const (
// exponent.
shardVsContainerExponent = shardWidthExponent - 16
// width of roaring containers is 2^16
containerWidth = 1 << 16
// snapshotExt is the file extension used for an in-process snapshot.
snapshotExt = ".snapshotting"
@ -612,8 +615,17 @@ func (f *fragment) value(columnID uint64, bitDepth uint) (value uint64, exists b
return value, true, nil
}
// clearValue uses a column of bits to clear a multi-bit value.
func (f *fragment) clearValue(columnID uint64, bitDepth uint, value uint64) (changed bool, err error) {
return f.setValueBase(columnID, bitDepth, value, true)
}
// setValue uses a column of bits to set a multi-bit value.
func (f *fragment) setValue(columnID uint64, bitDepth uint, value uint64) (changed bool, err error) {
return f.setValueBase(columnID, bitDepth, value, false)
}
func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value uint64, clear bool) (changed bool, err error) {
f.mu.Lock()
defer f.mu.Unlock()
@ -633,19 +645,26 @@ func (f *fragment) setValue(columnID uint64, bitDepth uint, value uint64) (chang
}
}
// Mark value as set.
if c, err := f.unprotectedSetBit(uint64(bitDepth), columnID); err != nil {
return changed, errors.Wrap(err, "marking not-null")
} else if c {
changed = true
// Mark value as set (or cleared).
if clear {
if c, err := f.unprotectedClearBit(uint64(bitDepth), columnID); err != nil {
return changed, errors.Wrap(err, "clearing not-null")
} else if c {
changed = true
}
} else {
if c, err := f.unprotectedSetBit(uint64(bitDepth), columnID); err != nil {
return changed, errors.Wrap(err, "marking not-null")
} else if c {
changed = true
}
}
return changed, nil
}
// importSetValue is a more efficient SetValue just for imports.
func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value uint64) (changed bool, err error) { // nolint: unparam
func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value uint64, clear bool) (changed bool, err error) { // nolint: unparam
for i := uint(0); i < bitDepth; i++ {
if value&(1<<i) != 0 {
bit, err := f.pos(uint64(i), columnID)
@ -673,12 +692,20 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value uint64)
// Mark value as set.
p, err := f.pos(uint64(bitDepth), columnID)
if err != nil {
return changed, errors.Wrap(err, "marking not-null")
return changed, errors.Wrap(err, "getting not-null pos")
}
if c, err := f.storage.Add(p); err != nil {
return changed, errors.Wrap(err, "adding to storage")
} else if c {
changed = true
if clear {
if c, err := f.storage.Remove(p); err != nil {
return changed, errors.Wrap(err, "removing not-null from storage")
} else if c {
changed = true
}
} else {
if c, err := f.storage.Add(p); err != nil {
return changed, errors.Wrap(err, "adding not-null to storage")
} else if c {
changed = true
}
}
return changed, nil
@ -688,12 +715,11 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value uint64)
// A bitmap can be passed in to optionally filter the computed columns.
func (f *fragment) sum(filter *Row, bitDepth uint) (sum, count uint64, err error) {
// Compute count based on the existence row.
row := f.row(uint64(bitDepth))
consider := f.row(uint64(bitDepth))
if filter != nil {
count = row.intersectionCount(filter)
} else {
count = row.Count()
consider = consider.Intersect(filter)
}
count = consider.Count()
// Compute the sum based on the bit count of each row multiplied by the
// place value of each row. For example, 10 bits in the 1's place plus
@ -705,11 +731,7 @@ func (f *fragment) sum(filter *Row, bitDepth uint) (sum, count uint64, err error
var cnt uint64
for i := uint(0); i < bitDepth; i++ {
row := f.row(uint64(i))
if filter != nil {
cnt = row.intersectionCount(filter)
} else {
cnt = row.Count()
}
cnt = row.intersectionCount(consider)
sum += (1 << i) * cnt
}
@ -1418,20 +1440,20 @@ func (f *fragment) mergeBlock(id int, data []pairSet) (sets, clears []pairSet, e
// bulkImport bulk imports a set of bits and then snapshots the storage.
// The cache is updated to reflect the new data.
func (f *fragment) bulkImport(rowIDs, columnIDs []uint64) error {
func (f *fragment) bulkImport(rowIDs, columnIDs []uint64, options *ImportOptions) error {
// Verify that there are an equal number of row ids and column ids.
if len(rowIDs) != len(columnIDs) {
return fmt.Errorf("mismatch of row/column len: %d != %d", len(rowIDs), len(columnIDs))
}
if f.mutexVector != nil {
if f.mutexVector != nil && !options.Clear {
return f.bulkImportMutex(rowIDs, columnIDs)
}
return f.bulkImportStandard(rowIDs, columnIDs)
return f.bulkImportStandard(rowIDs, columnIDs, options)
}
// bulkImportStandard performs a bulk import on a standard fragment.
func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64) error {
func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *ImportOptions) error {
// Create a temporary bitmap which will be populated by rowIDs and columnIDs
// and then merged into the existing fragment's bitmap.
localBitmap := roaring.NewBitmap()
@ -1480,10 +1502,18 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64) error {
// Merge localBitmap into fragment's existing data.
var results *roaring.Bitmap
if f.storage.Count() > 0 {
results = f.storage.Union(localBitmap)
if options.Clear {
if f.storage.Count() > 0 {
results = f.storage.Difference(localBitmap)
} else {
results = roaring.NewBitmap()
}
} else {
results = localBitmap
if f.storage.Count() > 0 {
results = f.storage.Union(localBitmap)
} else {
results = localBitmap
}
}
// Update cache counts for all affected rows.
@ -1589,7 +1619,7 @@ func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64) error {
}
// importValue bulk imports a set of range-encoded values.
func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint) error {
func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear bool) error {
f.mu.Lock()
defer f.mu.Unlock()
// Verify that there are an equal number of column ids and values.
@ -1604,7 +1634,7 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint) error
for i := range columnIDs {
columnID, value := columnIDs[i], values[i]
_, err := f.importSetValue(columnID, bitDepth, value)
_, err := f.importSetValue(columnID, bitDepth, value, clear)
if err != nil {
return errors.Wrap(err, "setting")
}
@ -1624,7 +1654,7 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint) error
// importRoaring imports from the official roaring data format defined at
// https://github.com/RoaringBitmap/RoaringFormatSpec or from pilosa's version
// of the roaring format. The cache is updated to reflect the new data.
func (f *fragment) importRoaring(data []byte) error {
func (f *fragment) importRoaring(data []byte, clear bool) error {
f.mu.Lock()
defer f.mu.Unlock()
bm := roaring.NewBitmap()
@ -1652,8 +1682,12 @@ func (f *fragment) importRoaring(data []byte) error {
lastRow = vRow
}
if f.storage.Count() > 0 {
bm = f.storage.Union(bm)
if clear {
bm = f.storage.Difference(bm)
} else {
if f.storage.Count() > 0 {
bm = f.storage.Union(bm)
}
}
for _, rowID := range rowSet {
@ -1957,15 +1991,82 @@ func (f *fragment) readCacheFromArchive(r io.Reader) error {
return nil
}
func (f *fragment) rows() []uint64 {
i, _ := f.storage.Containers.Iterator(0)
rows := make([]uint64, 0)
// rowFilter is a function signature for controlling iteration over containers
// in a fragment. It will be invoked on each container found and returns two
// booleans. The first is whether the row this container is in should be
// included or skipped, and the second is whether to stop processing or
// continue.
type rowFilter func(rowID, key uint64, c *roaring.Container) (include, done bool)
// filterWithLimit returns a filter which will only allow a limited number of
// rows to be returned. It should be applied last so that it is only called (and
// therefore only updates its internal state) if the row is being included by
// every other filter.
func filterWithLimit(limit uint64) rowFilter {
return func(rowID, key uint64, c *roaring.Container) (include, done bool) {
if limit > 0 {
limit--
return true, false
}
return false, true
}
}
func filterColumn(col uint64) rowFilter {
return func(rowID, key uint64, c *roaring.Container) (include, done bool) {
colID := col % ShardWidth
colKey := ((rowID * ShardWidth) + colID) >> 16
colVal := uint16(colID & 0xFFFF) // columnID within the container
return colKey == key && c.Contains(colVal), false
}
}
// TODO: this works, but it would be more performant if the fragment could seek
// to the next row in the rows list rather than asking the filter for each
// container serially. The container iterator would need to expose a seek
// method, and the rowFilter would need some way of communicating to
// fragment.rows what the next rowID to seek to is.
func filterWithRows(rows []uint64) rowFilter {
loc := 0
return func(rowID, key uint64, c *roaring.Container) (include, done bool) {
if loc >= len(rows) {
return false, true
}
i := sort.Search(len(rows[loc:]), func(i int) bool {
return rows[loc+i] >= rowID
})
loc += i
if loc >= len(rows) {
return false, true
}
if rows[loc] == rowID {
if loc == len(rows)-1 {
done = true
}
return true, done
}
return false, false
}
}
// rows returns all rows starting from 'start'. Filters will be applied in
// order. All filters must return true to include the row. Once a row is
// included, further containers in that row will be skipped. So, for a row to be
// included, there must be one container in that row where all filters return
// true. For a row to be skipped, at least one filter must return false for each
// container in that row (it need not be the same filter for each). Any filter
// returning done == true will cause processing to stop after all filters for
// this container have been processed. The rows accumulated up to this point
// (including this row if all filters passed) will be returned.
func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 {
startKey := rowToKey(start)
i, _ := f.storage.Containers.Iterator(startKey)
rows := make([]uint64, 0)
var lastRow uint64 = math.MaxUint64
// Loop over the existing containers.
for i.Next() {
key, _ := i.Value()
key, c := i.Value()
// virtual row for the current container
vRow := key >> shardVsContainerExponent
@ -1975,44 +2076,63 @@ func (f *fragment) rows() []uint64 {
continue
}
rows = append(rows, vRow)
lastRow = vRow
}
return rows
}
func (f *fragment) rowsForColumn(columnID uint64) []uint64 {
var colKey uint64
colID := columnID % ShardWidth
i, _ := f.storage.Containers.Iterator(0)
colVal := uint16(colID & 0xFFFF)
rows := make([]uint64, 0)
// Loop over the existing containers.
for i.Next() {
key, c := i.Value()
// virtual row for the current container
vRow := key >> shardVsContainerExponent
// column container key for virtual row
colKey = ((vRow * ShardWidth) + colID) >> 16
if colKey != key {
continue
// apply filters
addRow, done := true, false
for _, filter := range filters {
var d bool
addRow, d = filter(vRow, key, c)
done = done || d
if !addRow {
break
}
}
if c.Contains(colVal) {
if addRow {
lastRow = vRow
rows = append(rows, vRow)
}
if done {
return rows
}
}
return rows
}
type rowIterator struct {
f *fragment
rowIDs []uint64
cur int
wrap bool
}
func (f *fragment) rowIterator(wrap bool, filters ...rowFilter) *rowIterator {
return &rowIterator{
f: f,
rowIDs: f.rows(0, filters...), // TODO: this may be memory intensive in high cardinality cases
wrap: wrap,
}
}
func (ri *rowIterator) Seek(rowID uint64) {
idx := sort.Search(len(ri.rowIDs), func(i int) bool {
return ri.rowIDs[i] >= rowID
})
ri.cur = idx
}
func (ri *rowIterator) Next() (r *Row, rowID uint64, wrapped bool) {
if ri.cur >= len(ri.rowIDs) {
if !ri.wrap || len(ri.rowIDs) == 0 {
return nil, 0, true
}
ri.Seek(0)
wrapped = true
}
rowID = ri.rowIDs[ri.cur]
r = ri.f.row(rowID)
ri.cur += 1
return r, rowID, wrapped
}
// FragmentBlock represents info about a subsection of the rows in a block.
// This is used for comparing data in remote blocks for active anti-entropy.
type FragmentBlock struct {
@ -2292,7 +2412,7 @@ func newRowsVector(f *fragment) *rowsVector {
// Additionally, it returns true if a value was found,
// otherwise it returns false.
func (v *rowsVector) Get(colID uint64) (uint64, bool, error) {
rows := v.f.rowsForColumn(colID)
rows := v.f.rows(0, filterColumn(colID))
if len(rows) > 1 {
return 0, false, errors.New("found multiple row values for column")
} else if len(rows) == 1 {
@ -2301,6 +2421,13 @@ func (v *rowsVector) Get(colID uint64) (uint64, bool, error) {
return 0, false, nil
}
// rowToKey converts a Pilosa row ID to the key of the container which starts
// that row in the bitmap which represents this entire fragment. A fragment is
// all the rows within a shard within a field concatenated together.
func rowToKey(rowID uint64) (key uint64) {
return rowID * (ShardWidth / containerWidth)
}
// boolVector implements the vector interface by looking
// at data in rows 0 and 1.
type boolVector struct {
@ -2318,7 +2445,7 @@ func newBoolVector(f *fragment) *boolVector {
// Additionally, it returns true if a value was found,
// otherwise it returns false.
func (v *boolVector) Get(colID uint64) (uint64, bool, error) {
rows := v.f.rowsForColumn(colID)
rows := v.f.rows(0, filterColumn(colID))
if len(rows) > 1 {
return 0, false, errors.New("found multiple row values for column")
} else if len(rows) == 1 {

413
fragment_internal_test.go
View file

@ -228,6 +228,34 @@ func TestFragment_SetValue(t *testing.T) {
}
})
t.Run("Clear", func(t *testing.T) {
f := mustOpenFragment("i", "f", viewStandard, 0, "")
defer f.Close()
// Set value.
if changed, err := f.setValue(100, 16, 3829); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal("expected change")
}
// Clear value should overwrite all bits, and set not-null to 0.
if changed, err := f.clearValue(100, 16, 2028); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal("expected change")
}
// Read value.
if value, exists, err := f.value(100, 16); err != nil {
t.Fatal(err)
} else if value != 0 {
t.Fatalf("unexpected value: %d", value)
} else if exists {
t.Fatal("expected to not exist")
}
})
t.Run("NotExists", func(t *testing.T) {
f := mustOpenFragment("i", "f", viewStandard, 0, "")
defer f.Close()
@ -1253,12 +1281,15 @@ func TestFragment_SetMutex(t *testing.T) {
}
}
// Ensure a fragment can import mutually exclusive values.
func TestFragment_ImportMutex(t *testing.T) {
// Ensure a fragment can import into set fields.
func TestFragment_ImportSet(t *testing.T) {
tests := []struct {
rowIDs []uint64
colIDs []uint64
exp map[uint64][]uint64
setRowIDs []uint64
setColIDs []uint64
setExp map[uint64][]uint64
clearRowIDs []uint64
clearColIDs []uint64
clearExp map[uint64][]uint64
}{
{
[]uint64{1, 1, 1, 1},
@ -1266,6 +1297,129 @@ func TestFragment_ImportMutex(t *testing.T) {
map[uint64][]uint64{
1: {0, 1, 2, 3},
},
[]uint64{},
[]uint64{},
map[uint64][]uint64{
1: {0, 1, 2, 3},
},
},
{
[]uint64{1, 1, 1, 1, 2, 2, 2, 2},
[]uint64{0, 1, 2, 3, 0, 1, 2, 3},
map[uint64][]uint64{
1: {0, 1, 2, 3},
2: {0, 1, 2, 3},
},
[]uint64{1, 1, 2},
[]uint64{1, 2, 3},
map[uint64][]uint64{
1: {0, 3},
2: {0, 1, 2},
},
},
{
[]uint64{1, 1, 1, 1, 2},
[]uint64{0, 1, 2, 3, 1},
map[uint64][]uint64{
1: {0, 1, 2, 3},
2: {1},
},
[]uint64{1, 1, 1, 1},
[]uint64{0, 1, 2, 3},
map[uint64][]uint64{
1: {},
2: {1},
},
},
{
[]uint64{1, 1, 1, 1, 2, 2, 1},
[]uint64{0, 1, 2, 3, 1, 8, 1},
map[uint64][]uint64{
1: {0, 1, 2, 3},
2: {1, 8},
},
[]uint64{1, 1},
[]uint64{0, 0},
map[uint64][]uint64{
1: {1, 2, 3},
2: {1, 8},
},
},
{
[]uint64{1, 2, 3},
[]uint64{8, 8, 8},
map[uint64][]uint64{
1: {8},
2: {8},
3: {8},
},
[]uint64{1, 2, 3},
[]uint64{9, 9, 9},
map[uint64][]uint64{
1: {8},
2: {8},
3: {8},
},
},
}
for i, test := range tests {
t.Run(fmt.Sprintf("importset%d", i), func(t *testing.T) {
f := mustOpenFragment("i", "f", viewStandard, 0, "")
defer f.Close()
// Set import.
err := f.bulkImport(test.setRowIDs, test.setColIDs, &ImportOptions{})
if err != nil {
t.Fatalf("bulk importing ids: %v", err)
}
// Check for expected results.
for k, v := range test.setExp {
cols := f.row(k).Columns()
if !reflect.DeepEqual(cols, v) {
t.Fatalf("expected: %v, but got: %v", v, cols)
}
}
// Clear import.
err = f.bulkImport(test.clearRowIDs, test.clearColIDs, &ImportOptions{Clear: true})
if err != nil {
t.Fatalf("bulk clearing ids: %v", err)
}
// Check for expected results.
for k, v := range test.clearExp {
cols := f.row(k).Columns()
if !reflect.DeepEqual(cols, v) {
t.Fatalf("expected: %v, but got: %v", v, cols)
}
}
})
}
}
// Ensure a fragment can import mutually exclusive values.
func TestFragment_ImportMutex(t *testing.T) {
tests := []struct {
setRowIDs []uint64
setColIDs []uint64
setExp map[uint64][]uint64
clearRowIDs []uint64
clearColIDs []uint64
clearExp map[uint64][]uint64
}{
{
[]uint64{1, 1, 1, 1},
[]uint64{0, 1, 2, 3},
map[uint64][]uint64{
1: {0, 1, 2, 3},
},
[]uint64{},
[]uint64{},
map[uint64][]uint64{
1: {0, 1, 2, 3},
},
},
{
[]uint64{1, 1, 1, 1, 2, 2, 2, 2},
@ -1274,6 +1428,12 @@ func TestFragment_ImportMutex(t *testing.T) {
1: {},
2: {0, 1, 2, 3},
},
[]uint64{1, 1, 2},
[]uint64{1, 2, 3},
map[uint64][]uint64{
1: {},
2: {0, 1, 2},
},
},
{
[]uint64{1, 1, 1, 1, 2},
@ -1282,6 +1442,12 @@ func TestFragment_ImportMutex(t *testing.T) {
1: {0, 2, 3},
2: {1},
},
[]uint64{1, 1, 1, 1},
[]uint64{0, 1, 2, 3},
map[uint64][]uint64{
1: {},
2: {1},
},
},
{
[]uint64{1, 1, 1, 1, 2, 2, 1},
@ -1290,6 +1456,12 @@ func TestFragment_ImportMutex(t *testing.T) {
1: {0, 1, 2, 3},
2: {8},
},
[]uint64{1, 1},
[]uint64{0, 0},
map[uint64][]uint64{
1: {1, 2, 3},
2: {8},
},
},
{
[]uint64{1, 2, 3},
@ -1299,6 +1471,13 @@ func TestFragment_ImportMutex(t *testing.T) {
2: {},
3: {8},
},
[]uint64{1, 2, 3},
[]uint64{9, 9, 9},
map[uint64][]uint64{
1: {},
2: {},
3: {8},
},
},
}
@ -1307,13 +1486,28 @@ func TestFragment_ImportMutex(t *testing.T) {
f := mustOpenMutexFragment("i", "f", viewStandard, 0, "")
defer f.Close()
err := f.bulkImport(test.rowIDs, test.colIDs)
// Set import.
err := f.bulkImport(test.setRowIDs, test.setColIDs, &ImportOptions{})
if err != nil {
t.Fatalf("bulk importing ids: %v", err)
}
// Check for expected results.
for k, v := range test.exp {
for k, v := range test.setExp {
cols := f.row(k).Columns()
if !reflect.DeepEqual(cols, v) {
t.Fatalf("expected: %v, but got: %v", v, cols)
}
}
// Clear import.
err = f.bulkImport(test.clearRowIDs, test.clearColIDs, &ImportOptions{Clear: true})
if err != nil {
t.Fatalf("bulk clearing ids: %v", err)
}
// Check for expected results.
for k, v := range test.clearExp {
cols := f.row(k).Columns()
if !reflect.DeepEqual(cols, v) {
t.Fatalf("expected: %v, but got: %v", v, cols)
@ -1326,9 +1520,12 @@ func TestFragment_ImportMutex(t *testing.T) {
// Ensure a fragment can import bool values.
func TestFragment_ImportBool(t *testing.T) {
tests := []struct {
rowIDs []uint64
colIDs []uint64
exp map[uint64][]uint64
setRowIDs []uint64
setColIDs []uint64
setExp map[uint64][]uint64
clearRowIDs []uint64
clearColIDs []uint64
clearExp map[uint64][]uint64
}{
{
[]uint64{1, 1, 1, 1},
@ -1336,6 +1533,11 @@ func TestFragment_ImportBool(t *testing.T) {
map[uint64][]uint64{
1: {0, 1, 2, 3},
},
[]uint64{},
[]uint64{},
map[uint64][]uint64{
1: {0, 1, 2, 3},
},
},
{
[]uint64{0, 0, 0, 0, 1, 1, 1, 1},
@ -1344,6 +1546,13 @@ func TestFragment_ImportBool(t *testing.T) {
0: {},
1: {0, 1, 2, 3},
},
[]uint64{1, 1, 2},
[]uint64{1, 2, 3},
map[uint64][]uint64{
0: {},
1: {0, 3},
2: {},
},
},
{
[]uint64{0, 0, 0, 0, 1},
@ -1352,6 +1561,12 @@ func TestFragment_ImportBool(t *testing.T) {
0: {0, 2, 3},
1: {1},
},
[]uint64{1, 1, 1, 1},
[]uint64{0, 1, 2, 3},
map[uint64][]uint64{
0: {0, 2, 3},
1: {},
},
},
{
[]uint64{1, 1, 1, 1, 0, 0, 1},
@ -1360,6 +1575,12 @@ func TestFragment_ImportBool(t *testing.T) {
0: {8},
1: {0, 1, 2, 3},
},
[]uint64{1, 1},
[]uint64{0, 0},
map[uint64][]uint64{
0: {8},
1: {1, 2, 3},
},
},
{
[]uint64{0, 1, 2},
@ -1369,6 +1590,13 @@ func TestFragment_ImportBool(t *testing.T) {
1: {}, // This isn't {8} because fragment doesn't validate bool values.
2: {8},
},
[]uint64{1, 2, 3},
[]uint64{9, 9, 9},
map[uint64][]uint64{
0: {},
1: {},
2: {8},
},
},
}
@ -1377,13 +1605,28 @@ func TestFragment_ImportBool(t *testing.T) {
f := mustOpenBoolFragment("i", "f", viewStandard, 0, "")
defer f.Close()
err := f.bulkImport(test.rowIDs, test.colIDs)
// Set import.
err := f.bulkImport(test.setRowIDs, test.setColIDs, &ImportOptions{})
if err != nil {
t.Fatalf("bulk importing ids: %v", err)
}
// Check for expected results.
for k, v := range test.exp {
for k, v := range test.setExp {
cols := f.row(k).Columns()
if !reflect.DeepEqual(cols, v) {
t.Fatalf("expected: %v, but got: %v", v, cols)
}
}
// Clear import.
err = f.bulkImport(test.clearRowIDs, test.clearColIDs, &ImportOptions{Clear: true})
if err != nil {
t.Fatalf("bulk importing ids: %v", err)
}
// Check for expected results.
for k, v := range test.clearExp {
cols := f.row(k).Columns()
if !reflect.DeepEqual(cols, v) {
t.Fatalf("expected: %v, but got: %v", v, cols)
@ -1427,6 +1670,7 @@ func BenchmarkFragment_FullSnapshot(b *testing.B) {
rows := make([]uint64, sz)
cols := make([]uint64, sz)
options := &ImportOptions{}
max := 0
for row := 0; row < 100; row++ {
val := 1
@ -1437,7 +1681,7 @@ func BenchmarkFragment_FullSnapshot(b *testing.B) {
val += 2
i++
}
if err := f.bulkImport(rows, cols); err != nil {
if err := f.bulkImport(rows, cols, options); err != nil {
b.Fatalf("Error Building Sample: %s", err)
}
if row > max {
@ -1477,8 +1721,9 @@ func BenchmarkFragment_Import(b *testing.B) {
}
b.ResetTimer()
b.ReportAllocs()
options := &ImportOptions{}
for i := 0; i < b.N; i++ {
if err := f.bulkImport(rows, cols); err != nil {
if err := f.bulkImport(rows, cols, options); err != nil {
b.Fatalf("Error Building Sample: %s", err)
}
}
@ -1563,12 +1808,12 @@ func TestFragment_RowsIteration(t *testing.T) {
}
}
ids := f.rows()
ids := f.rows(0)
if !reflect.DeepEqual(expectedAll, ids) {
t.Fatalf("Do not match %v %v", expectedAll, ids)
}
ids = f.rowsForColumn(1)
ids = f.rows(0, filterColumn(1))
if !reflect.DeepEqual(expectedOdd, ids) {
t.Fatalf("Do not match %v %v", expectedOdd, ids)
}
@ -1587,12 +1832,12 @@ func TestFragment_RowsIteration(t *testing.T) {
t.Fatal(err)
}
ids := f.rows()
ids := f.rows(0)
if !reflect.DeepEqual(expected, ids) {
t.Fatalf("Do not match %v %v", expected, ids)
}
ids = f.rowsForColumn(66000)
ids = f.rows(0, filterColumn(66000))
if !reflect.DeepEqual(expected, ids) {
t.Fatalf("Do not match %v %v", expected, ids)
}
@ -1610,11 +1855,11 @@ func TestFragment_RowsIteration(t *testing.T) {
t.Fatal(err)
}
ids := f.rows()
ids := f.rows(0)
if !reflect.DeepEqual(expectedRows, ids) {
t.Fatalf("Do not match %v %v", expectedRows, ids)
}
ids = f.rowsForColumn(c)
ids = f.rows(0, filterColumn(c))
if !reflect.DeepEqual(expectedRows, ids) {
t.Fatalf("Do not match %v %v", expectedRows, ids)
}
@ -1655,7 +1900,7 @@ func TestFragment_RoaringImport(t *testing.T) {
if err != nil {
t.Fatalf("writing to buffer: %v", err)
}
f.importRoaring(buf.Bytes())
f.importRoaring(buf.Bytes(), false)
exp := calcExpected(test[:num+1]...)
for row, expCols := range exp {
cols := f.row(uint64(row)).Columns()
@ -1692,7 +1937,8 @@ func TestFragment_RoaringImportTopN(t *testing.T) {
f := mustOpenFragment("i", "f", viewStandard, 0, CacheTypeRanked)
defer f.Close()
err := f.bulkImport(test.rowIDs, test.colIDs)
options := &ImportOptions{}
err := f.bulkImport(test.rowIDs, test.colIDs, options)
if err != nil {
t.Fatalf("bulk importing ids: %v", err)
}
@ -1705,7 +1951,7 @@ func TestFragment_RoaringImportTopN(t *testing.T) {
t.Fatalf("post bulk import:\n exp: %v\n got: %v\n", expPairs, pairs)
}
err = f.bulkImport(test.rowIDs2, test.colIDs2)
err = f.bulkImport(test.rowIDs2, test.colIDs2, options)
if err != nil {
t.Fatalf("bulk importing ids: %v", err)
}
@ -1726,7 +1972,7 @@ func TestFragment_RoaringImportTopN(t *testing.T) {
if err != nil {
t.Fatalf("writing to buffer: %v", err)
}
f.importRoaring(buf.Bytes())
f.importRoaring(buf.Bytes(), false)
rows, cols := toRowsCols(test.roaring)
expPairs = calcTop(append(test.rowIDs, rows...), append(test.colIDs, cols...))
pairs, err = f.top(topOptions{})
@ -1819,3 +2065,122 @@ func calcExpected(inputs ...[]uint64) [][]uint64 {
return ret
}
func TestFragmentRowIterator(t *testing.T) {
t.Run("basic", func(t *testing.T) {
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
f.mustSetBits(0, 0)
f.mustSetBits(1, 0)
f.mustSetBits(2, 0)
f.mustSetBits(3, 0)
iter := f.rowIterator(false)
for i := uint64(0); i < 4; i++ {
row, id, wrapped := iter.Next()
if id != i {
t.Fatalf("expected row %d but got %d", i, id)
}
if wrapped {
t.Fatalf("shouldn't have wrapped")
}
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
}
}
row, id, wrapped := iter.Next()
if row != nil {
t.Fatalf("row should be nil after iterator is exhausted, got %v", row.Columns())
}
if id != 0 {
t.Fatalf("id should be 0 after iterator is exhausted, got %d", id)
}
if !wrapped {
t.Fatalf("wrapped should be true after iterator is exhausted")
}
f.Close()
})
t.Run("skipped rows", func(t *testing.T) {
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
f.mustSetBits(1, 0)
f.mustSetBits(3, 0)
f.mustSetBits(5, 0)
f.mustSetBits(7, 0)
iter := f.rowIterator(false)
for i := uint64(1); i < 8; i += 2 {
row, id, wrapped := iter.Next()
if id != i {
t.Fatalf("expected row %d but got %d", i, id)
}
if wrapped {
t.Fatalf("shouldn't have wrapped")
}
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
}
}
row, id, wrapped := iter.Next()
if row != nil {
t.Fatalf("row should be nil after iterator is exhausted, got %v", row.Columns())
}
if id != 0 {
t.Fatalf("id should be 0 after iterator is exhausted, got %d", id)
}
if !wrapped {
t.Fatalf("wrapped should be true after iterator is exhausted")
}
f.Close()
})
t.Run("basic wrapped", func(t *testing.T) {
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
f.mustSetBits(0, 0)
f.mustSetBits(1, 0)
f.mustSetBits(2, 0)
f.mustSetBits(3, 0)
iter := f.rowIterator(true)
for i := uint64(0); i < 5; i++ {
row, id, wrapped := iter.Next()
if id != i%4 {
t.Fatalf("expected row %d but got %d", i%4, id)
}
if wrapped && i < 4 {
t.Fatalf("shouldn't have wrapped")
} else if !wrapped && i >= 4 {
t.Fatalf("should have wrapped")
}
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
}
}
f.Close()
})
t.Run("skipped rows wrapped", func(t *testing.T) {
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
f.mustSetBits(1, 0)
f.mustSetBits(3, 0)
f.mustSetBits(5, 0)
f.mustSetBits(7, 0)
iter := f.rowIterator(true)
for i := uint64(1); i < 10; i += 2 {
row, id, wrapped := iter.Next()
if id != i%8 {
t.Errorf("expected row %d but got %d", i%8, id)
}
if wrapped && i < 8 {
t.Errorf("shouldn't have wrapped")
} else if !wrapped && i >= 8 {
t.Errorf("should have wrapped")
}
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
}
}
f.Close()
})
}

81
http/client.go
View file

@ -294,13 +294,22 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index s
}
// 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) error {
func (c *InternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []pilosa.Bit, opts ...pilosa.ImportOption) error {
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)
@ -314,7 +323,7 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard
// Import to each node.
for _, node := range nodes {
if err := c.importNode(ctx, node, index, field, buf); err != nil {
if err := c.importNode(ctx, node, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
}
}
@ -332,13 +341,22 @@ func getCoordinatorNode(nodes []*pilosa.Node) *pilosa.Node {
}
// ImportK bulk imports bits specified by string keys to a host.
func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits []pilosa.Bit) error {
func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits []pilosa.Bit, opts ...pilosa.ImportOption) error {
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)
@ -356,7 +374,7 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits
}
// Import to node.
if err := c.importNode(ctx, coord, index, field, buf); err != nil {
if err := c.importNode(ctx, coord, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err)
}
@ -410,11 +428,18 @@ func (c *InternalClient) marshalImportPayload(index, field string, shard uint64,
}
// importNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte) error {
func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte, opts *pilosa.ImportOptions) error {
// Create URL & HTTP request.
path := fmt.Sprintf("/index/%s/field/%s/import", index, field)
u := nodePathToURL(node, path)
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
vals := url.Values{}
if opts.Clear {
vals.Set("clear", "true")
}
url := fmt.Sprintf("%s?%s", u.String(), vals.Encode())
req, err := http.NewRequest("POST", url, bytes.NewReader(buf))
if err != nil {
return errors.Wrap(err, "creating request")
}
@ -449,13 +474,22 @@ func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, inde
}
// 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) error {
func (c *InternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error {
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.marshalImportValuePayload(index, field, shard, vals)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
@ -469,7 +503,7 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
// Import to each node.
for _, node := range nodes {
if err := c.importNode(ctx, node, index, field, buf); err != nil {
if err := c.importNode(ctx, node, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
}
}
@ -478,12 +512,21 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
}
// ImportValueK bulk imports keyed field values to a host.
func (c *InternalClient) ImportValueK(ctx context.Context, index, field string, vals []pilosa.FieldValue) error {
func (c *InternalClient) ImportValueK(ctx context.Context, index, field string, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error {
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)
if err != nil {
return errors.Wrap(err, "applying option")
}
}
// Get the coordinator node; all bits are sent to the
// primary translate store (i.e. coordinator).
nodes, err := c.Nodes(ctx)
@ -496,7 +539,7 @@ func (c *InternalClient) ImportValueK(ctx context.Context, index, field string,
}
// Import to node.
if err := c.importNode(ctx, coord, index, field, buf); err != nil {
if err := c.importNode(ctx, coord, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err)
}
@ -527,7 +570,7 @@ func (c *InternalClient) marshalImportValuePayload(index, field string, shard ui
// ImportRoaring does fast import of raw bits in roaring format (pilosa or
// official format, see API.ImportRoaring).
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, remote bool, data []byte) error {
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, remote bool, data []byte, opts ...pilosa.ImportOption) error {
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
@ -537,7 +580,21 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, ind
uri = c.defaultURI
}
url := fmt.Sprintf("%s/index/%s/field/%s/import-roaring/%d?remote=%v", uri, index, field, shard, remote)
// Set up import options.
options := &pilosa.ImportOptions{}
for _, opt := range opts {
err := opt(options)
if err != nil {
return errors.Wrap(err, "applying option")
}
}
vals := url.Values{}
vals.Set("remote", strconv.FormatBool(remote))
if options.Clear {
vals.Set("clear", "true")
}
url := fmt.Sprintf("%s/index/%s/field/%s/import-roaring/%d?%s", uri, index, field, shard, vals.Encode())
// Generate HTTP request.
req, err := http.NewRequest("POST", url, bytes.NewBuffer(data))

160
http/client_test.go
View file

@ -362,6 +362,22 @@ func TestClient_Import(t *testing.T) {
if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{6}) {
t.Fatalf("unexpected columns: %+v", a)
}
// 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)
}
// 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)
}
}
// Ensure client can bulk import data.
@ -392,13 +408,13 @@ func TestClient_ImportRoaring(t *testing.T) {
// Send import request.
host := cluster[0].URL()
c := MustNewClient(host, http.GetHTTPClient(nil))
roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100")
roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100") // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringData); err != nil {
t.Fatal(err)
}
hldr := test.Holder{Holder: cluster[0].Server.Holder()}
// Verify data.
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected columns: %+v", a)
}
@ -407,13 +423,79 @@ func TestClient_ImportRoaring(t *testing.T) {
}
hldr2 := test.Holder{Holder: cluster[1].Server.Holder()}
// Verify data.
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected columns: %+v", a)
}
// Ensure that sending a roaring import with the clear flag works as expected.
roaringDataClear, _ := hex.DecodeString("3A30000001000000010001001000000003000400") // [65539, 65540]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringDataClear, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Ensure that sending a roaring import with the clear flag works as expected.
roaringDataClear, _ = hex.DecodeString("3A300000020000000000010001000100180000001C0000000400060001000300") // [4, 6, 65537, 65539]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringDataClear, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 5, 7, 8, 9, 10}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 5, 7, 8, 9, 10}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Ensure that sending a roaring import with the clear flag works as expected.
roaringDataClear, _ = hex.DecodeString("3B3001000100000900010000000100010009000100") // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringDataClear, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
}
// Ensure client can bulk import data.
@ -624,7 +706,7 @@ func TestClient_ImportKeys(t *testing.T) {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=50, cnt=3", sum, cnt)
}
// Verify Range
// Verify Range.
queryRequest := &pilosa.QueryRequest{
Query: fmt.Sprintf(`Range(%s>10)`, fldName),
Remote: false,
@ -638,6 +720,37 @@ func TestClient_ImportKeys(t *testing.T) {
if !reflect.DeepEqual(result.Results[0].(*pilosa.Row).Keys, []string{"col2", "col3"}) {
t.Fatalf("unexpected column keys: %s", spew.Sdump(result))
}
// Clear data.
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnKey: "col2", Value: 20},
}, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err = field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 30 || cnt != 2 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=30, cnt=2", sum, cnt)
}
// Verify Range.
queryRequest = &pilosa.QueryRequest{
Query: fmt.Sprintf(`Range(%s>10)`, fldName),
Remote: false,
}
result, err = c.Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(result.Results[0].(*pilosa.Row).Keys, []string{"col3"}) {
t.Fatalf("unexpected column keys: %s", spew.Sdump(result))
}
})
}
@ -706,6 +819,45 @@ func TestClient_ImportValue(t *testing.T) {
if max != 40 || cnt != 1 {
t.Fatalf("unexpected values: got max=%v, count=%v; expected max=40, cnt=1", max, cnt)
}
// 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 {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err = field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 20 || cnt != 1 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=20, cnt=1", sum, cnt)
}
// Verify Min with Filter.
filter, err = field.Range(fldName, pql.GT, 40)
if err != nil {
t.Fatal(err)
}
min, cnt, err = field.Min(filter, fldName)
if err != nil {
t.Fatal(err)
}
if min != -100 || cnt != 0 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=-100, cnt=0", min, cnt)
}
// Verify Max.
max, cnt, err = field.Max(nil, fldName)
if err != nil {
t.Fatal(err)
}
if max != 20 || cnt != 1 {
t.Fatalf("unexpected values: got max=%v, count=%v; expected max=20, cnt=1", max, cnt)
}
}
// Ensure client can bulk import data while tracking existence.

17
http/handler.go
View file

@ -181,8 +181,8 @@ func (h *Handler) populateValidators() {
h.validators["DeleteIndex"] = queryValidationSpecRequired()
h.validators["PostField"] = queryValidationSpecRequired()
h.validators["DeleteField"] = queryValidationSpecRequired()
h.validators["PostImport"] = queryValidationSpecRequired()
h.validators["PostImportRoaring"] = queryValidationSpecRequired().Optional("remote")
h.validators["PostImport"] = queryValidationSpecRequired().Optional("clear")
h.validators["PostImportRoaring"] = queryValidationSpecRequired().Optional("remote", "clear")
h.validators["PostQuery"] = queryValidationSpecRequired().Optional("shards", "columnAttrs", "excludeRowAttrs", "excludeColumns")
h.validators["GetInfo"] = queryValidationSpecRequired()
h.validators["RecalculateCaches"] = queryValidationSpecRequired()
@ -994,6 +994,10 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
indexName := mux.Vars(r)["index"]
fieldName := mux.Vars(r)["field"]
// If the clear flag is true, treat the import as clear bits.
q := r.URL.Query()
doClear := q.Get("clear") == "true"
// Get index and field type to determine how to handle the
// import data.
field, err := h.api.Field(r.Context(), indexName, fieldName)
@ -1026,7 +1030,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
return
}
if err := h.api.ImportValue(r.Context(), req); err != nil {
if err := h.api.ImportValue(r.Context(), req, pilosa.OptImportOptionsClear(doClear)); err != nil {
switch errors.Cause(err) {
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)
@ -1044,7 +1048,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
return
}
if err := h.api.Import(r.Context(), req); err != nil {
if err := h.api.Import(r.Context(), req, pilosa.OptImportOptionsClear(doClear)); err != nil {
switch errors.Cause(err) {
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)
@ -1507,6 +1511,9 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
remote = true
}
// If the clear flag is true, treat the import as clear bits.
doClear := q.Get("clear") == "true"
// Read entire body.
body, err := ioutil.ReadAll(r.Body)
if err != nil {
@ -1523,7 +1530,7 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
resp := &pilosa.ImportResponse{}
// TODO give meaningful stats for import
err = h.api.ImportRoaring(r.Context(), urlVars["index"], urlVars["field"], shard, remote, body)
err = h.api.ImportRoaring(r.Context(), urlVars["index"], urlVars["field"], shard, remote, body, pilosa.OptImportOptionsClear(doClear))
if err != nil {
resp.Err = err.Error()
if _, ok := err.(pilosa.BadRequestError); ok {

1025
internal/public.pb.go
View file

File diff suppressed because it is too large Load diff

20
internal/public.proto
View file

@ -8,12 +8,27 @@ message Row {
repeated Attr Attrs = 2;
}
message RowIdentifiers {
repeated uint64 Rows = 1;
repeated string Keys = 2;
}
message Pair {
uint64 ID = 1;
string Key = 3;
uint64 Count = 2;
}
message FieldRow{
string Field = 1;
uint64 RowID = 2;
}
message GroupCount{
repeated FieldRow Group = 1;
uint64 Count = 2;
}
message ValCount {
int64 Val = 1;
int64 Count = 2;
@ -64,8 +79,11 @@ message QueryResult {
Row Row = 1;
uint64 N = 2;
repeated Pair Pairs = 3;
ValCount ValCount = 5;
bool Changed = 4;
ValCount ValCount = 5;
repeated uint64 RowIDs = 7;
repeated GroupCount GroupCounts = 8;
RowIdentifiers RowIdentifiers = 9;
}
message ImportRequest {

4
pilosa.go
View file

@ -62,7 +62,9 @@ var (
ErrNodeNotCoordinator = errors.New("node is not the coordinator")
ErrResizeNotRunning = errors.New("no resize job currently running")
ErrNotImplemented = errors.New("not implemented")
ErrNotImplemented = errors.New("not implemented")
ErrFieldsArgumentRequired = errors.New("fields argument required")
ErrExpectedFieldListArgument = errors.New("expected field list argument")
)
// apiMethodNotAllowedError wraps an error value indicating that a particular

73
test/pilosa.go
View file

@ -269,6 +269,79 @@ func (m *Command) RecalculateCaches() error {
// Cluster represents a Pilosa cluster (multiple Command instances)
type Cluster []*Command
// Query executes an API.Query through one of the cluster's node's API. It fails
// the test if there is an error.
func (c Cluster) Query(t testing.TB, index, query string) pilosa.QueryResponse {
if len(c) == 0 {
t.Fatal("must have at least one node in cluster to query")
}
return c[0].MustQuery(t, &pilosa.QueryRequest{Index: index, Query: query})
}
func (c Cluster) ImportBits(t testing.TB, index, field string, rowcols [][2]uint64) {
byShard := make(map[uint64][][2]uint64)
for _, rowcol := range rowcols {
shard := rowcol[1] / pilosa.ShardWidth
byShard[shard] = append(byShard[shard], rowcol)
}
for shard, bits := range byShard {
rowIDs := make([]uint64, len(bits))
colIDs := make([]uint64, len(bits))
for i, bit := range bits {
rowIDs[i] = bit[0]
colIDs[i] = bit[1]
}
nodes, err := c[0].API.ShardNodes(context.Background(), index, shard)
if err != nil {
t.Fatalf("getting shard nodes: %v", err)
}
// TODO won't be necessary to do all nodes once that works hits
for _, node := range nodes {
for _, com := range c {
if com.API.Node().ID != node.ID {
continue
}
err := com.API.Import(context.Background(), &pilosa.ImportRequest{
Index: index,
Field: field,
Shard: shard,
RowIDs: rowIDs,
ColumnIDs: colIDs,
})
if err != nil {
t.Fatalf("importing data: %v", err)
}
}
}
}
}
// CreateField creates the index (if necessary) and field specified.
func (c Cluster) CreateField(t testing.TB, index string, iopts pilosa.IndexOptions, field string, fopts ...pilosa.FieldOption) *pilosa.Field {
idx, err := c[0].API.CreateIndex(context.Background(), index, iopts)
if err != nil && !strings.Contains(err.Error(), "index already exists") {
t.Fatalf("creating index: %v", err)
} else if err != nil { // index exists
idx, err = c[0].API.Index(context.Background(), index)
if err != nil {
t.Fatalf("getting index: %v", err)
}
}
if idx.Options() != iopts {
t.Logf("existing index options:\n%v\ndon't match given opts:\n%v\n in pilosa/test.Cluster.CreateField", idx.Options(), iopts)
}
f, err := c[0].API.CreateField(context.Background(), index, field, fopts...)
// we'll assume the field doesn't exist because checking if the options
// match seems painful.
if err != nil {
t.Fatalf("creating field: %v", err)
}
return f
}
// Start runs a Cluster
func (c Cluster) Start() error {
var gossipSeeds = make([]string, len(c))