From a91014c7bb7765a69dd06fdc668122de0a4f4727 Mon Sep 17 00:00:00 2001 From: Travis Date: Tue, 5 May 2020 23:30:26 -0500 Subject: [PATCH] add FieldValue call --- apimethod_string.go | 8 +++- executor.go | 91 +++++++++++++++++++++++++++++++++++++++++++- executor_test.go | 92 +++++++++++++++++++++++++++++++++++++++++++++ pql/ast.go | 7 ++++ server/grpc.go | 92 +++++++++++++++++++++++++++++++++------------ 5 files changed, 262 insertions(+), 28 deletions(-) diff --git a/apimethod_string.go b/apimethod_string.go index d8217b035..b694fcb9b 100644 --- a/apimethod_string.go +++ b/apimethod_string.go @@ -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) { diff --git a/executor.go b/executor.go index 3badd4906..b58266fb9 100644 --- a/executor.go +++ b/executor.go @@ -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 diff --git a/executor_test.go b/executor_test.go index f894a10ee..781fb83e3 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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) { diff --git a/pql/ast.go b/pql/ast.go index 355b97740..e4163a66e 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -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{}{ diff --git a/server/grpc.go b/server/grpc.go index 37059983f..156574a1b 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -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})