mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
Merge branch 'develop' into core-structs
This commit is contained in:
Matt Jaffee
commit
27c071ac85
3 changed files with 105 additions and 102 deletions
129
executor.go
129
executor.go
|
|
@ -183,9 +183,7 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
|
|||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
return e.executeCount(ctx, index, c, shards, opt)
|
||||
case "Set":
|
||||
return e.executeSetBit(ctx, index, c, opt)
|
||||
case "SetValue":
|
||||
return nil, e.executeSetValue(ctx, index, c, opt)
|
||||
return e.executeSet(ctx, index, c, opt)
|
||||
case "SetRowAttrs":
|
||||
return nil, e.executeSetRowAttrs(ctx, index, c, opt)
|
||||
case "SetColumnAttrs":
|
||||
|
|
@ -1059,8 +1057,8 @@ func (e *executor) executeClearBitField(ctx context.Context, index string, c *pq
|
|||
return ret, nil
|
||||
}
|
||||
|
||||
// executeSetBit executes a Set() call.
|
||||
func (e *executor) executeSetBit(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
|
||||
// executeSet executes a Set() call.
|
||||
func (e *executor) executeSet(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
|
||||
fieldName, err := c.FieldArg()
|
||||
if err != nil {
|
||||
return false, errors.New("Set() argument required: field")
|
||||
|
|
@ -1076,14 +1074,7 @@ func (e *executor) executeSetBit(ctx context.Context, index string, c *pql.Call,
|
|||
return false, ErrFieldNotFound
|
||||
}
|
||||
|
||||
// Read fields using labels.
|
||||
rowID, ok, err := c.UintArg(fieldName)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("reading Set() row: %v", err)
|
||||
} else if !ok {
|
||||
return false, fmt.Errorf("Set() row argument '%v' required", rowLabel)
|
||||
}
|
||||
|
||||
// Read colID using labels.
|
||||
colID, ok, err := c.UintArg("_" + columnLabel)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("reading Set() column: %v", err)
|
||||
|
|
@ -1091,20 +1082,40 @@ func (e *executor) executeSetBit(ctx context.Context, index string, c *pql.Call,
|
|||
return false, fmt.Errorf("Set() column argument '%v' required", columnLabel)
|
||||
}
|
||||
|
||||
var timestamp *time.Time
|
||||
sTimestamp, ok := c.Args["_timestamp"].(string)
|
||||
if ok {
|
||||
t, err := time.Parse(TimeFormat, sTimestamp)
|
||||
if f.Type() == FieldTypeInt {
|
||||
// Read remaining fields using labels.
|
||||
rowVal, ok, err := c.IntArg(fieldName)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("invalid date: %s", sTimestamp)
|
||||
return false, fmt.Errorf("reading Set() row: %v", err)
|
||||
} else if !ok {
|
||||
return false, fmt.Errorf("Set() row argument '%v' required", rowLabel)
|
||||
}
|
||||
timestamp = &t
|
||||
}
|
||||
|
||||
return e.executeSetBitField(ctx, index, c, f, colID, rowID, timestamp, opt)
|
||||
return e.executeSetValueField(ctx, index, c, f, colID, rowVal, opt)
|
||||
} else {
|
||||
// Read remaining fields using labels.
|
||||
rowID, ok, err := c.UintArg(fieldName)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("reading Set() row: %v", err)
|
||||
} else if !ok {
|
||||
return false, fmt.Errorf("Set() row argument '%v' required", rowLabel)
|
||||
}
|
||||
|
||||
var timestamp *time.Time
|
||||
sTimestamp, ok := c.Args["_timestamp"].(string)
|
||||
if ok {
|
||||
t, err := time.Parse(TimeFormat, sTimestamp)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("invalid date: %s", sTimestamp)
|
||||
}
|
||||
timestamp = &t
|
||||
}
|
||||
|
||||
return e.executeSetBitField(ctx, index, c, f, colID, rowID, timestamp, opt)
|
||||
}
|
||||
}
|
||||
|
||||
// executeSetBitField executes a Set() call for a specific view.
|
||||
// executeSetBitField executes a Set() call for a specific field.
|
||||
func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *execOptions) (bool, error) {
|
||||
shard := colID / ShardWidth
|
||||
ret := false
|
||||
|
|
@ -1136,64 +1147,36 @@ func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.
|
|||
return ret, nil
|
||||
}
|
||||
|
||||
// executeSetValue executes a SetValue() call.
|
||||
func (e *executor) executeSetValue(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
|
||||
// Parse labels.
|
||||
columnID, ok, err := c.UintArg(columnLabel)
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading SetValue() column: %v", err)
|
||||
} else if !ok {
|
||||
return fmt.Errorf("SetValue() column field '%v' required", columnLabel)
|
||||
}
|
||||
// executeSetValueField executes a Set() call for a specific int field.
|
||||
func (e *executor) executeSetValueField(ctx context.Context, index string, c *pql.Call, f *Field, colID uint64, value int64, opt *execOptions) (bool, error) {
|
||||
shard := colID / ShardWidth
|
||||
ret := false
|
||||
|
||||
// Copy args and remove reserved fields.
|
||||
args := pql.CopyArgs(c.Args)
|
||||
// While field could technically work as a ColumnAttr argument, we are treating it as a reserved word primarily to avoid confusion.
|
||||
// Also, if we ever need to make ColumnAttrs field-specific, then having this reserved word prevents backward incompatibility.
|
||||
delete(args, columnLabel)
|
||||
|
||||
// Set values.
|
||||
for name, value := range args {
|
||||
// Retrieve field.
|
||||
field := e.Holder.Field(index, name)
|
||||
if field == nil {
|
||||
return ErrFieldNotFound
|
||||
}
|
||||
|
||||
switch value := value.(type) {
|
||||
case int64:
|
||||
if _, err := field.SetValue(columnID, value); err != nil {
|
||||
return err
|
||||
for _, node := range e.Cluster.shardNodes(index, shard) {
|
||||
// Update locally if host matches.
|
||||
if node.ID == e.Node.ID {
|
||||
val, err := f.SetValue(colID, value)
|
||||
if err != nil {
|
||||
return false, err
|
||||
} else if val {
|
||||
ret = true
|
||||
}
|
||||
default:
|
||||
return ErrInvalidBSIGroupValueType
|
||||
continue
|
||||
}
|
||||
field.Stats.Count("SetValue", 1, 1.0)
|
||||
}
|
||||
|
||||
// Do not forward call if this is already being forwarded.
|
||||
if opt.Remote {
|
||||
return nil
|
||||
}
|
||||
// Do not forward call if this is already being forwarded.
|
||||
if opt.Remote {
|
||||
continue
|
||||
}
|
||||
|
||||
// Execute on remote nodes in parallel.
|
||||
nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID)
|
||||
resp := make(chan error, len(nodes))
|
||||
for _, node := range nodes {
|
||||
go func(node *Node) {
|
||||
_, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, opt)
|
||||
resp <- err
|
||||
}(node)
|
||||
}
|
||||
|
||||
// Return first error.
|
||||
for range nodes {
|
||||
if err := <-resp; err != nil {
|
||||
return err
|
||||
// Forward call to remote node otherwise.
|
||||
if res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, opt); err != nil {
|
||||
return false, err
|
||||
} else {
|
||||
ret = res[0].(bool)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
// executeSetRowAttrs executes a SetRowAttrs() call.
|
||||
|
|
|
|||
|
|
@ -405,9 +405,9 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
|
|||
}
|
||||
|
||||
// Set bsiGroup values.
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetValue(col=10, f=25)`}); err != nil {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(10, f=25)`}); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetValue(col=100, f=10)`}); err != nil {
|
||||
} else if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(100, f=10)`}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -440,19 +440,19 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
|
|||
}
|
||||
|
||||
t.Run("ErrColumnBSIGroupRequired", func(t *testing.T) {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetValue(invalid_column_name=10, f=100)`}); err == nil || errors.Cause(err).Error() != `SetValue() column field 'col' required` {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(invalid_column_name=10, f=100)`}); err == nil || errors.Cause(err).Error() != `field not found` {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("ErrColumnBSIGroupValue", func(t *testing.T) {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetValue(invalid_column_name="bad_column", f=100)`}); err == nil || errors.Cause(err).Error() != `SetValue() column field 'col' required` {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("bad_column", f=100)`}); err == nil || errors.Cause(err).Error() != `string 'col' value not allowed unless index 'keys' option enabled` {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("ErrInvalidBSIGroupValueType", func(t *testing.T) {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetValue(col=10, f="hello")`}); err == nil || errors.Cause(err) != pilosa.ErrInvalidBSIGroupValueType {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(10, f="hello")`}); err == nil || errors.Cause(err).Error() != `string 'row' value not allowed unless field 'keys' option enabled` {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
})
|
||||
|
|
@ -748,14 +748,14 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
|
|||
Set(1, x=1)
|
||||
Set(` + strconv.Itoa(ShardWidth+2) + `, x=2)
|
||||
|
||||
SetValue(col=0, f=20)
|
||||
SetValue(col=1, f=-5)
|
||||
SetValue(col=2, f=-5)
|
||||
SetValue(col=3, f=10)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth) + `, f=30)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth+2) + `, f=40)
|
||||
SetValue(col=` + strconv.Itoa((5*ShardWidth)+100) + `, f=50)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth+1) + `, f=60)
|
||||
Set(0, f=20)
|
||||
Set(1, f=-5)
|
||||
Set(2, f=-5)
|
||||
Set(3, f=10)
|
||||
Set(` + strconv.Itoa(ShardWidth) + `, f=30)
|
||||
Set(` + strconv.Itoa(ShardWidth+2) + `, f=40)
|
||||
Set(` + strconv.Itoa((5*ShardWidth)+100) + `, f=50)
|
||||
Set(` + strconv.Itoa(ShardWidth+1) + `, f=60)
|
||||
`}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -844,13 +844,13 @@ func TestExecutor_Execute_Sum(t *testing.T) {
|
|||
Set(0, x=0)
|
||||
Set(` + strconv.Itoa(ShardWidth+1) + `, x=0)
|
||||
|
||||
SetValue(col=0, foo=20)
|
||||
SetValue(col=0, bar=2000)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth) + `, foo=30)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth+2) + `, foo=40)
|
||||
SetValue(col=` + strconv.Itoa((5*ShardWidth)+100) + `, foo=50)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth+1) + `, foo=60)
|
||||
SetValue(col=0, other=1000)
|
||||
Set(0, foo=20)
|
||||
Set(0, bar=2000)
|
||||
Set(` + strconv.Itoa(ShardWidth) + `, foo=30)
|
||||
Set(` + strconv.Itoa(ShardWidth+2) + `, foo=40)
|
||||
Set(` + strconv.Itoa((5*ShardWidth)+100) + `, foo=50)
|
||||
Set(` + strconv.Itoa(ShardWidth+1) + `, foo=60)
|
||||
Set(0, other=1000)
|
||||
`}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -959,15 +959,15 @@ func TestExecutor_Execute_BSIGroupRange(t *testing.T) {
|
|||
Set(0, f=0)
|
||||
Set(` + strconv.Itoa(ShardWidth+1) + `, f=0)
|
||||
|
||||
SetValue(col=50, foo=20)
|
||||
SetValue(col=50, bar=2000)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth) + `, foo=30)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth+2) + `, foo=10)
|
||||
SetValue(col=` + strconv.Itoa((5*ShardWidth)+100) + `, foo=20)
|
||||
SetValue(col=` + strconv.Itoa(ShardWidth+1) + `, foo=60)
|
||||
SetValue(col=0, other=1000)
|
||||
SetValue(col=0, edge=100)
|
||||
SetValue(col=1, edge=-100)
|
||||
Set(50, foo=20)
|
||||
Set(50, bar=2000)
|
||||
Set(` + strconv.Itoa(ShardWidth) + `, foo=30)
|
||||
Set(` + strconv.Itoa(ShardWidth+2) + `, foo=10)
|
||||
Set(` + strconv.Itoa((5*ShardWidth)+100) + `, foo=20)
|
||||
Set(` + strconv.Itoa(ShardWidth+1) + `, foo=60)
|
||||
Set(0, other=1000)
|
||||
Set(0, edge=100)
|
||||
Set(1, edge=-100)
|
||||
`}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
|
|||
20
pql/ast.go
20
pql/ast.go
|
|
@ -285,6 +285,26 @@ func (c *Call) UintArg(key string) (uint64, bool, error) {
|
|||
}
|
||||
}
|
||||
|
||||
// IntArg is for reading the value at key from call.Args as an int64. If the
|
||||
// key is not in Call.Args, the value of the returned bool will be false, and
|
||||
// the error will be nil. The value is assumed to be a unt64 or an int64 and
|
||||
// then cast to an int64. An error is returned if the value is not an int64 or
|
||||
// uint64.
|
||||
func (c *Call) IntArg(key string) (int64, bool, error) {
|
||||
val, ok := c.Args[key]
|
||||
if !ok {
|
||||
return 0, false, nil
|
||||
}
|
||||
switch tval := val.(type) {
|
||||
case int64:
|
||||
return tval, true, nil
|
||||
case uint64:
|
||||
return int64(tval), true, nil
|
||||
default:
|
||||
return 0, true, fmt.Errorf("could not convert %v of type %T to int64 in Call.IntArg", tval, tval)
|
||||
}
|
||||
}
|
||||
|
||||
// UintSliceArg reads the value at key from call.Args as a slice of uint64. If
|
||||
// the key is not in Call.Args, the value of the returned bool will be false,
|
||||
// and the error will be nil. If the value is a slice of int64 it will convert
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue