scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 08:10:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #337 from travisturner/fieldvalue

Browse source 
add FieldValue call
This commit is contained in:
Travis Turner 2020-05-06 12:29:39 -05:00 • committed by GitHub
commit 994ac5aad8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 262 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
apimethod_string.go
View file

@ -34,11 +34,15 @@ func _() {
_ = x[apiShardNodes-23]
_ = x[apiViews-24]
_ = x[apiApplySchema-25]
_ = x[apiStartTransaction-26]
_ = x[apiFinishTransaction-27]
_ = x[apiTransactions-28]
_ = x[apiGetTransaction-29]
}
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchema"
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransaction"
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 367}
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 367, 386, 406, 421, 438}
func (i apiMethod) String() string {
if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) {

91
executor.go
View file

@ -552,6 +552,9 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
return e.executeOptionsCall(ctx, index, c, shards, opt)
case "IncludesColumn":
return e.executeIncludesColumnCall(ctx, index, c, shards, opt)
case "FieldValue":
statFn()
return e.executeFieldValueCall(ctx, index, c, shards, opt)
case "All":
statFn()
return e.executeAllCall(ctx, index, c, shards, opt)
@ -663,6 +666,92 @@ func (e *executor) executeIncludesColumnCall(ctx context.Context, index string,
return result.(bool), nil
}
// executeFieldValueCall executes a FieldValue() call.
func (e *executor) executeFieldValueCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
fieldName, ok := c.Args["field"].(string)
if !ok || fieldName == "" {
return ValCount{}, errors.New("FieldValue(): field required")
}
// Fetch field.
field := e.Holder.Field(index, fieldName)
if field == nil {
return ValCount{}, ErrFieldNotFound
}
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
return ValCount{}, ErrIndexNotFound
}
var colID uint64
if colKey, ok := c.Args["column"].(string); ok && idx.Keys() {
if id, err := e.Cluster.translateIndexKey(ctx, index, colKey); err != nil {
return ValCount{}, errors.Wrap(err, "getting column id")
} else {
colID = id
}
} else {
if id, ok, err := c.UintArg("column"); !ok || err != nil {
// TODO: this error is getting swallowed somewhere (via curl)
return ValCount{}, errors.Wrap(err, "getting column argument")
} else {
colID = id
}
}
shard := colID / ShardWidth
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeFieldValueCallShard(ctx, field, colID, shard)
}
// Select single returned result at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
other, _ := prev.(ValCount)
if other.Count == 1 {
return other
}
return v
}
result, err := e.mapReduce(ctx, index, []uint64{shard}, c, opt, mapFn, reduceFn)
if err != nil {
return ValCount{}, errors.Wrap(err, "map reduce")
}
other, _ := result.(ValCount)
return other, nil
}
func (e *executor) executeFieldValueCallShard(ctx context.Context, field *Field, col uint64, shard uint64) (ValCount, error) {
value, exists, err := field.Value(col)
if err != nil {
return ValCount{}, errors.Wrap(err, "getting field value")
} else if !exists {
return ValCount{}, nil
}
other := ValCount{
Count: 1,
}
if field.Type() == FieldTypeInt {
other.Val = value
} else if field.Type() == FieldTypeDecimal {
other.DecimalVal = &pql.Decimal{
Value: value,
Scale: field.Options().Scale}
other.FloatVal = 0
other.Val = 0
}
return other, nil
}
// executeAllCall executes an All() call.
func (e *executor) executeAllCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
rslt := NewRow()
@ -2266,7 +2355,7 @@ func (e *executor) executeRowShard(ctx context.Context, index string, c *pql.Cal
return e.executeRowBSIGroupShard(ctx, index, c, shard)
}
// Fetch column label from index.
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
return nil, ErrIndexNotFound

92
executor_test.go
View file

@ -3272,6 +3272,98 @@ func TestExecutor_Execute_Not(t *testing.T) {
})
}
// Ensure an all query can be executed.
func TestExecutor_Execute_FieldValue(t *testing.T) {
c := test.MustRunCluster(t, 2)
defer c.Close()
//hldr := test.Holder{Holder: c[0].Server.Holder()}
node0 := c[0]
node1 := c[1]
// Index with IDs
c.CreateField(t, "i", pilosa.IndexOptions{Keys: false}, "f", pilosa.OptFieldTypeInt(-1100, 1000))
c.CreateField(t, "i", pilosa.IndexOptions{Keys: false}, "dec", pilosa.OptFieldTypeDecimal(3))
if _, err := node0.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `
Set(1, f=3)
Set(2, f=-4)
Set(` + strconv.Itoa(ShardWidth+1) + `, f=3)
Set(1, dec=12.985)
Set(2, dec=-4.234)
`}); err != nil {
t.Fatal(err)
}
// Index with Keys
c.CreateField(t, "ik", pilosa.IndexOptions{Keys: true}, "f", pilosa.OptFieldTypeInt(-1100, 1000))
c.CreateField(t, "ik", pilosa.IndexOptions{Keys: true}, "dec", pilosa.OptFieldTypeDecimal(3))
if _, err := node0.API.Query(context.Background(), &pilosa.QueryRequest{Index: "ik", Query: `
Set("one", f=3)
Set("two", f=-4)
Set("one", dec=12.985)
Set("two", dec=-4.234)
`}); err != nil {
t.Fatal(err)
}
tests := []struct {
index string
qry string
expVal interface{}
expErr string
}{
// IDs
{index: "i", qry: "FieldValue(field=f, column=1)", expVal: int64(3)},
{index: "i", qry: "FieldValue(field=f, column=2)", expVal: int64(-4)},
{index: "i", qry: "FieldValue(field=f, column=" + strconv.Itoa(ShardWidth+1) + ")", expVal: int64(3)},
{index: "i", qry: "FieldValue(field=dec, column=1)", expVal: pql.NewDecimal(12985, 3)},
{index: "i", qry: "FieldValue(field=dec, column=2)", expVal: pql.NewDecimal(-4234, 3)},
// Keys
{index: "ik", qry: "FieldValue(field=f, column='one')", expVal: int64(3)},
{index: "ik", qry: "FieldValue(field=f, column='two')", expVal: int64(-4)},
{index: "ik", qry: "FieldValue(field=dec, column='one')", expVal: pql.NewDecimal(12985, 3)},
{index: "ik", qry: "FieldValue(field=dec, column='two')", expVal: pql.NewDecimal(-4234, 3)},
// Errors
{index: "i", qry: "FieldValue()", expErr: pilosa.ErrFieldRequired.Error()},
}
for n, node := range []*test.Command{node0, node1} {
for i, test := range tests {
if res, err := node.API.Query(context.Background(), &pilosa.QueryRequest{Index: test.index, Query: test.qry}); err != nil && test.expErr == "" {
t.Fatal(err)
} else if err != nil && test.expErr != "" {
if !strings.Contains(err.Error(), test.expErr) {
t.Fatalf("test %d on node%d expected error: %s, but got: %s", i, n, test.expErr, err)
}
} else if err == nil && test.expErr != "" {
t.Fatalf("test %d on node%d expected error but got nil", i, n)
} else if vc, ok := res.Results[0].(pilosa.ValCount); !ok {
t.Fatalf("test %d on node%d expected pilosa.ValCount, but got: %T", i, n, res.Results[0])
} else if vc.Count != 1 {
t.Fatalf("test %d on node%d expected Count 1, but got: %d", i, n, vc.Count)
} else {
switch exp := test.expVal.(type) {
case pql.Decimal:
if *vc.DecimalVal != exp {
t.Fatalf("test %d on node%d expected pql.Decimal(%s), but got: %s", i, n, exp, vc.DecimalVal)
}
case int64:
if vc.Val != exp {
t.Fatalf("test %d on node%d expected int64(%d), but got: %d", i, n, exp, vc.Val)
}
default:
t.Fatalf("test %d on node%d received unhandled type: %T", i, n, test.expVal)
}
}
}
}
}
// Ensure an all query can be executed.
func TestExecutor_Execute_All(t *testing.T) {
t.Run("ColumnID", func(t *testing.T) {

7
pql/ast.go
View file

@ -349,6 +349,13 @@ var callInfoByFunc = map[string]callInfo{
"Difference": {allowUnknown: false},
"Intersect": {allowUnknown: false},
"Not": {allowUnknown: false},
"FieldValue": {
allowUnknown: false,
prototypes: map[string]interface{}{
"field": "",
"column": stringOrInt64,
},
},
"All": {
allowUnknown: false,
prototypes: map[string]interface{}{

92
server/grpc.go
View file

@ -408,11 +408,19 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
var err error
if fi := field.ForeignIndex(); fi != "" {
// Get the value from the int field.
intVal, ok, err := field.Value(col)
pql := fmt.Sprintf("FieldValue(field=%s, column=%d)", field.Name(), col)
query := pilosa.QueryRequest{
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "getting int value")
} else if ok {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(intVal)})
return errors.Wrap(err, "getting int field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
@ -435,12 +443,20 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
value, exists, err := field.Value(col)
pql := fmt.Sprintf("FieldValue(field=%s, column=%d)", field.Name(), col)
query := pilosa.QueryRequest{
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "getting int field value for column")
} else if exists {
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: value}})
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -448,12 +464,20 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
}
case "decimal":
dec, exists, err := field.DecimalValue(col)
pql := fmt.Sprintf("FieldValue(field=%s, column=%d)", field.Name(), col)
query := pilosa.QueryRequest{
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "getting decimal field value for column")
} else if exists {
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: dec.Value, Scale: dec.Scale}}})
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -633,11 +657,19 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
var err error
if fi := field.ForeignIndex(); fi != "" {
// Get the value from the int field.
intVal, ok, err := field.Value(id)
pql := fmt.Sprintf("FieldValue(field=%s, column=%d)", field.Name(), id)
query := pilosa.QueryRequest{
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "getting int value")
} else if ok {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(intVal)})
return errors.Wrap(err, "getting int field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
@ -660,12 +692,20 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
value, exists, err := field.Value(id)
pql := fmt.Sprintf("FieldValue(field=%s, column=%d)", field.Name(), id)
query := pilosa.QueryRequest{
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "getting int field value for column")
} else if exists {
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: value}})
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -673,18 +713,20 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
}
case "decimal":
// Translate column key.
id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), col)
if err != nil {
return errors.Wrap(err, "translating column key")
pql := fmt.Sprintf("FieldValue(field=%s, column='%s')", field.Name(), col)
query := pilosa.QueryRequest{
Index: req.Index,
Query: pql,
}
dec, exists, err := field.DecimalValue(id)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "getting decimal field value for column")
} else if exists {
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: dec.Value, Scale: dec.Scale}}})
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})