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

Merge pull request #1699 from travisturner/import-clear

Browse source 
add `clear` functional option for imports
This commit is contained in:
Travis Turner 2018-10-24 16:15:49 -05:00 • committed by GitHub
commit 79bf01f79a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 785 additions and 146 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.

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
}

97
fragment.go
View file

@ -612,8 +612,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 +642,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 +689,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 +712,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 +728,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 +1437,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 +1499,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 +1616,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 +1631,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 +1651,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 +1679,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 {

282
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)
}
}
@ -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{})

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 {