From a7a15c64a21ff2d54c6d4bd7b8e6feb4ddaeff23 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 22 Oct 2018 15:59:24 -0500 Subject: [PATCH] support clear imports to int fields. fix bug in fragment.sum --- api.go | 21 +++++++++--- client.go | 8 ++--- ctl/import.go | 2 +- ctl/import_test.go | 69 +++++++++++++++++++++++++++------------ field.go | 4 +-- fragment.go | 69 +++++++++++++++++++++++++-------------- fragment_internal_test.go | 28 ++++++++++++++++ http/client.go | 28 +++++++++++----- http/handler.go | 2 +- 9 files changed, 165 insertions(+), 66 deletions(-) diff --git a/api.go b/api.go index bb900db4e..a81f1db12 100644 --- a/api.go +++ b/api.go @@ -757,11 +757,20 @@ func (api *API) Import(_ context.Context, req *ImportRequest, opts ...ImportOpti } // ImportValue bulk imports values into a particular field. -func (api *API) ImportValue(_ context.Context, req *ImportValueRequest) 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 := &ImportOptions{} + for _, opt := range opts { + err := opt(options) + if err != nil { + return errors.Wrap(err, "applying option") + } + } + index := api.holder.Index(req.Index) if index == nil { return newNotFoundError(ErrIndexNotFound) @@ -783,13 +792,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 889544557..e6a02aa96 100644 --- a/client.go +++ b/client.go @@ -42,8 +42,8 @@ type InternalClient interface { 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 @@ -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/ctl/import.go b/ctl/import.go index 8f0f4a4e5..d49b50eb3 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -380,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 fdf73720a..ad4fd6cf0 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -103,29 +103,58 @@ func TestImportCommand_Basic(t *testing.T) { // 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/field.go b/field.go index 1a880a266..3422e0e19 100644 --- a/field.go +++ b/field.go @@ -1126,7 +1126,7 @@ func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time, opts } // 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) @@ -1174,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 } } diff --git a/fragment.go b/fragment.go index 94e45ab1e..d45757762 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<