diff --git a/executor.go b/executor.go index 399da685d..b39e26f58 100644 --- a/executor.go +++ b/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. diff --git a/executor_test.go b/executor_test.go index 6ab7472b1..41341b814 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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) } diff --git a/pql/ast.go b/pql/ast.go index 0bcc582d4..dc16f4cdf 100644 --- a/pql/ast.go +++ b/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