diff --git a/api.go b/api.go index 0b27d19a8..af206496b 100644 --- a/api.go +++ b/api.go @@ -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) } diff --git a/client.go b/client.go index a2ca0bf96..f6c49d14b 100644 --- a/client.go +++ b/client.go @@ -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 { diff --git a/cmd/import.go b/cmd/import.go index 7bfca1a84..878aad286 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -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 diff --git a/cmd/import_test.go b/cmd/import_test.go index ba015dc04..7b566ce65 100644 --- a/cmd/import_test.go +++ b/cmd/import_test.go @@ -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) } diff --git a/ctl/import.go b/ctl/import.go index d2694a64f..d49b50eb3 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -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") } } diff --git a/ctl/import_test.go b/ctl/import_test.go index 2913005fb..ad4fd6cf0 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -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. diff --git a/docs/administration.md b/docs/administration.md index e626cc0e5..3f07a5b68 100644 --- a/docs/administration.md +++ b/docs/administration.md @@ -82,6 +82,18 @@ For example, importing a file with the following contents will result in columns

Note that you must first create a field. View Create Field for more details. The `-e` flag can create the necessary schema when using a field of type "set".

+#### 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. diff --git a/field.go b/field.go index 63bb08b63..bc656fa54 100644 --- a/field.go +++ b/field.go @@ -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 } diff --git a/fragment.go b/fragment.go index 63c9cce37..10e5e3920 100644 --- a/fragment.go +++ b/fragment.go @@ -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< 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 { diff --git a/fragment_internal_test.go b/fragment_internal_test.go index be0db3b01..ac080b5b7 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -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{}) diff --git a/http/client.go b/http/client.go index 922ac6234..06bb31479 100644 --- a/http/client.go +++ b/http/client.go @@ -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)) diff --git a/http/client_test.go b/http/client_test.go index accbf86e4..23d8b8573 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -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. diff --git a/http/handler.go b/http/handler.go index 8b239ec2d..6007d72a8 100644 --- a/http/handler.go +++ b/http/handler.go @@ -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 {