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

support clear imports to int fields. fix bug in fragment.sum

Browse source
This commit is contained in:
Travis Turner 2018-10-22 15:59:24 -05:00
parent caf8e06712
commit a7a15c64a2
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
9 changed files with 165 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
api.go
View file

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

8
client.go
View file

@ -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 {

2
ctl/import.go
View file

@ -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")
}
}

69
ctl/import_test.go
View file

@ -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.

4
field.go
View file

@ -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
}
}

69
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
}
@ -1425,7 +1444,7 @@ func (f *fragment) bulkImport(rowIDs, columnIDs []uint64, options *ImportOptions
}
if f.mutexVector != nil && !options.Clear {
return f.bulkImportMutex(rowIDs, columnIDs, options)
return f.bulkImportMutex(rowIDs, columnIDs)
}
return f.bulkImportStandard(rowIDs, columnIDs, options)
}
@ -1508,7 +1527,7 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor
// mutex restrictions. Because the mutex requirements must be checked
// against storage, this method must acquire a write lock on the fragment
// during the entire process, and it handles every bit independently.
func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64, options *ImportOptions) error {
func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64) error {
f.mu.Lock()
defer f.mu.Unlock()
@ -1597,7 +1616,7 @@ func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64, options *ImportOp
}
// 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.
@ -1612,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")
}

28
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()

28
http/client.go
View file

@ -473,13 +473,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)
@ -491,9 +500,6 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
return fmt.Errorf("shard nodes: %s", err)
}
// Set up import options.
options := &pilosa.ImportOptions{}
// Import to each node.
for _, node := range nodes {
if err := c.importNode(ctx, node, index, field, buf, options); err != nil {
@ -505,12 +511,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)
@ -522,9 +537,6 @@ func (c *InternalClient) ImportValueK(ctx context.Context, index, field string,
return fmt.Errorf("could not find the coordinator node")
}
// Set up import options.
options := &pilosa.ImportOptions{}
// Import to node.
if err := c.importNode(ctx, coord, index, field, buf, options); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err)

2
http/handler.go
View file

@ -1030,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)