Merge pull request #1711 from molecula/seebs/clusterIngest

CORE-826: cluster support for ingest API
This commit is contained in:
seebs 2021-09-27 16:22:37 -05:00 committed by GitHub
commit 8fd6ebc8da
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 4335 additions and 819 deletions

110
api.go
View file

@ -1826,6 +1826,44 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
return nil
}
// ingestNodeOperationsForFields does the actual work of applying operations
// to a given index with a map of known fields and an already-parsed
// ShardedRequest. This is used locally on the node that first receives
// the request, after it does the parsing, and on other nodes because the
// format they get is already that rather than JSON, so it's the common
// path *after* key translation and sorting into shards.
func (api *API) ingestNodeOperationsForFields(ctx context.Context, qcx *Qcx, index *Index, knownFields map[string]*Field, req *ingest.ShardedRequest) error {
eg, ctx := errgroup.WithContext(ctx)
for shard, ops := range req.Ops {
// create new local copies of these values so the goroutine uses these
// copies, and doesn't read the actual loop variables, which are being
// changed by the loop.
shard, ops := shard, ops
eg.Go(func() error {
return api.applyOperations(ctx, qcx, index, shard, knownFields, ops)
})
}
return eg.Wait()
}
// IngestNodeOperations handles protobuf-formatted data which does not need
// key translation and is applicable to this specific node.
func (api *API) IngestNodeOperations(ctx context.Context, qcx *Qcx, indexName string, req *ingest.ShardedRequest) error {
index := api.holder.Index(indexName)
if index == nil {
api.server.logger.Errorf("ingest: no such index %q", indexName)
return newNotFoundError(ErrIndexNotFound, indexName)
}
fields := index.Fields()
knownFields := map[string]*Field{}
for _, field := range fields {
knownFields[field.name] = field
}
return api.ingestNodeOperationsForFields(ctx, qcx, index, knownFields, req)
}
// IngestOperations handles JSON-formatted data which may need key translation
// and may be for any or all nodes.
func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string, stream io.Reader) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.IngestOperations")
defer span.Finish()
@ -1841,34 +1879,32 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string
return newNotFoundError(ErrIndexNotFound, indexName)
}
fields := index.Fields()
var lookup ingest.KeyLookupFunc
var indexKeys ingest.KeyTranslator
if index.Keys() {
lookup = func(keys ...string) (map[string]uint64, error) {
return api.cluster.createIndexKeys(ctx, indexName, keys...)
}
indexKeys = newIngestKeyTranslatorFromCluster(ctx, api.cluster, indexName)
}
codec, err := ingest.NewJSONCodec(lookup)
codec, err := ingest.NewJSONCodec(indexKeys)
if err != nil {
return errors.Wrap(err, "creating JSON codec")
}
knownFields := map[string]*Field{}
for _, field := range fields {
var lookup ingest.KeyLookupFunc
var keys ingest.KeyTranslator
if field.usesKeys {
lookup = field.translateStore.CreateKeys
keys = newIngestKeyTranslatorFromStore(field.translateStore)
}
knownFields[field.name] = field
switch field.Type() {
case "set":
if err = codec.AddSetField(field.name, lookup); err != nil {
if err = codec.AddSetField(field.name, keys); err != nil {
return fmt.Errorf("adding set field to codec: %w", err)
}
case "time":
if err = codec.AddTimeQuantumField(field.name, lookup); err != nil {
if err = codec.AddTimeQuantumField(field.name, keys); err != nil {
return fmt.Errorf("adding time quantum field to codec: %w", err)
}
case "mutex":
if err = codec.AddMutexField(field.name, lookup); err != nil {
if err = codec.AddMutexField(field.name, keys); err != nil {
return fmt.Errorf("adding mutex field to codec: %w", err)
}
case "bool":
@ -1876,7 +1912,7 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string
return fmt.Errorf("adding bool field to codec: %w", err)
}
case "int":
if err = codec.AddIntField(field.name, lookup); err != nil {
if err = codec.AddIntField(field.name, keys); err != nil {
return fmt.Errorf("adding int field to codec: %w", err)
}
case "decimal":
@ -1896,17 +1932,53 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string
if err != nil {
return errors.Wrap(err, "parsing input data")
}
sharded, err := req.ByShard()
sharded, err := codec.RequestByShard(req)
if err != nil {
return errors.Wrap(err, "sharding input data")
}
eg, ctx := errgroup.WithContext(ctx)
// now that we have this, let's assign the shards to nodes
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
// oh hey an easy case: we're presumably the only node
if len(snap.Nodes) == 1 {
return api.ingestNodeOperationsForFields(ctx, qcx, index, knownFields, sharded)
}
// Created new ShardedRequest objects for every node, giving each of them
// all the shards that apply to them.
byNode := make(map[string]*ingest.ShardedRequest)
for shard, ops := range sharded.Ops {
// loop variable shadow capture is the go equivalent of man door hook hand
shard, ops := shard, ops
eg.Go(func() error {
return api.applyOperations(ctx, qcx, index, shard, knownFields, ops)
})
nodes := snap.ShardNodes(indexName, shard)
for _, node := range nodes {
forThisShard := byNode[node.ID]
if forThisShard == nil {
// Create new ShardedRequest for the target node, with its op map
// mapping this shard to the ops for this shard.
byNode[node.ID] = &ingest.ShardedRequest{Ops: map[uint64][]*ingest.Operation{shard: ops}}
continue
}
// Add this shard to the existing ShardedRequest's Ops map. Note that
// we don't have to worry about overwrites; we can't have seen this
// shard before, because we're in a range loop on a map where the shard
// is the key.
forThisShard.Ops[shard] = ops
}
}
eg, ctx := errgroup.WithContext(ctx)
for _, node := range snap.Nodes {
node := node
sharded := byNode[node.ID]
// Sometimes, there's nothing for a specific node.
if sharded == nil {
continue
}
if node.ID == api.NodeID() {
eg.Go(func() error {
return api.ingestNodeOperationsForFields(ctx, qcx, index, knownFields, sharded)
})
} else {
eg.Go(func() error {
return api.server.defaultClient.IngestNodeOperations(ctx, &node.URI, indexName, sharded)
})
}
}
return eg.Wait()
}
@ -3090,6 +3162,7 @@ const (
apiIDReset
apiPartitionNodes
apiIngestOperations
apiIngestNodeOperations
apiMutexCheck
)
@ -3159,5 +3232,6 @@ var methodsNormal = map[apiMethod]struct{}{
apiIDReset: {},
apiPartitionNodes: {},
apiIngestOperations: {},
apiIngestNodeOperations: {},
apiMutexCheck: {},
}

View file

@ -19,6 +19,7 @@ import (
"io"
"time"
"github.com/molecula/featurebase/v2/ingest"
pnet "github.com/molecula/featurebase/v2/net"
"github.com/molecula/featurebase/v2/topology"
)
@ -81,6 +82,7 @@ type InternalClient interface {
ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error)
MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error)
IngestNodeOperations(ctx context.Context, uri *pnet.URI, indexName string, ireq *ingest.ShardedRequest) error
IDAllocDataReader(ctx context.Context) (io.ReadCloser, error)
IndexTranslateDataReader(ctx context.Context, index string, partitionID int) (io.ReadCloser, error)
@ -218,6 +220,10 @@ func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index,
return nil, nil
}
func (n nopInternalClient) IngestNodeOperations(ctx context.Context, uri *pnet.URI, indexName string, ireq *ingest.ShardedRequest) error {
return nil
}
func (n nopInternalClient) ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error) {
return nil, nil
}

View file

@ -24,6 +24,7 @@ import (
"time"
"github.com/molecula/featurebase/v2/disco"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/logger"
"github.com/molecula/featurebase/v2/roaring"
"github.com/molecula/featurebase/v2/topology"
@ -1566,6 +1567,38 @@ func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys
return ids, nil
}
// This implements ingest's key translator interface on a cluster/index pair.
type clusterKeyTranslator struct {
ctx context.Context // we're created within a request context and need to pass that to cluster ops
c *cluster
indexName string
}
var _ ingest.KeyTranslator = &clusterKeyTranslator{}
func (i clusterKeyTranslator) TranslateKeys(keys ...string) (map[string]uint64, error) {
return i.c.createIndexKeys(i.ctx, i.indexName, keys...)
}
func (i clusterKeyTranslator) TranslateIDs(ids ...uint64) (map[uint64]string, error) {
keys, err := i.c.translateIndexIDs(i.ctx, i.indexName, ids)
if err != nil {
return nil, err
}
if len(keys) != len(ids) {
return nil, fmt.Errorf("translating %d id(s), got %d key(s)", len(ids), len(keys))
}
out := make(map[uint64]string, len(keys))
for i, id := range ids {
out[id] = keys[i]
}
return out, nil
}
func newIngestKeyTranslatorFromCluster(ctx context.Context, c *cluster, indexName string) *clusterKeyTranslator {
return &clusterKeyTranslator{ctx: ctx, c: c, indexName: indexName}
}
// TODO: remove this when it is no longer used
func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, keySet map[string]struct{}, writable bool) (map[string]uint64, error) {
keys := make([]string, 0, len(keySet))

View file

@ -21,6 +21,7 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/disco"
"github.com/molecula/featurebase/v2/ingest"
pnet "github.com/molecula/featurebase/v2/net"
"github.com/molecula/featurebase/v2/pb"
"github.com/molecula/featurebase/v2/pql"
@ -324,7 +325,18 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
}
decodeResizeAbortMessage(msg, mt)
return nil
case *ingest.ShardedRequest:
msg := &pb.ShardedIngestRequest{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshalling ShardedRequest")
}
req, err := s.decodeShardedIngestRequest(msg)
if err != nil {
return err
}
*mt = *req
return nil
default:
panic(fmt.Sprintf("unhandled pilosa.Message of type %T: %#v", mt, m))
}
@ -398,6 +410,8 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
return s.encodeResizeNodeMessage(mt)
case *pilosa.ResizeAbortMessage:
return s.encodeResizeAbortMessage(mt)
case *ingest.ShardedRequest:
return s.encodeShardedIngestRequest(mt)
}
return nil
}
@ -931,6 +945,48 @@ func (s Serializer) encodeTransactionStats(stats pilosa.TransactionStats) *pb.Tr
return &pb.TransactionStats{}
}
func (s Serializer) encodeShardedIngestRequest(req *ingest.ShardedRequest) *pb.ShardedIngestRequest {
if req == nil || len(req.Ops) == 0 {
return &pb.ShardedIngestRequest{}
}
out := &pb.ShardedIngestRequest{Ops: make(map[uint64]*pb.ShardIngestOperations, len(req.Ops))}
for shard, ops := range req.Ops {
out.Ops[shard] = s.encodeShardIngestOperations(ops)
}
return out
}
func (s Serializer) encodeShardIngestOperations(ops []*ingest.Operation) *pb.ShardIngestOperations {
out := &pb.ShardIngestOperations{}
for _, op := range ops {
if op == nil {
continue
}
out.Ops = append(out.Ops, s.encodeShardIngestOperation(op))
}
return out
}
func (s Serializer) encodeShardIngestOperation(op *ingest.Operation) *pb.ShardIngestOperation {
out := &pb.ShardIngestOperation{
OpType: op.OpType.String(),
ClearRecordIDs: op.ClearRecordIDs,
ClearFields: op.ClearFields,
FieldOps: make(map[string]*pb.FieldOperation, len(op.FieldOps)),
}
for k, v := range op.FieldOps {
if v == nil {
continue
}
out.FieldOps[k] = &pb.FieldOperation{
RecordIDs: v.RecordIDs,
Values: v.Values,
Signed: v.Signed,
}
}
return out
}
func (s Serializer) decodeResizeInstruction(ri *pb.ResizeInstruction, m *pilosa.ResizeInstruction) {
m.JobID = ri.JobID
m.Node = &topology.Node{}
@ -1829,3 +1885,57 @@ func decodeResizeNodeMessage(pb *pb.ResizeNodeMessage, m *pilosa.ResizeNodeMessa
func decodeResizeAbortMessage(pb *pb.ResizeAbortMessage, m *pilosa.ResizeAbortMessage) {
}
func (s Serializer) decodeShardedIngestRequest(req *pb.ShardedIngestRequest) (*ingest.ShardedRequest, error) {
if req == nil || len(req.Ops) == 0 {
return &ingest.ShardedRequest{}, nil
}
out := &ingest.ShardedRequest{Ops: make(map[uint64][]*ingest.Operation, len(req.Ops))}
for shard, ops := range req.Ops {
var err error
out.Ops[shard], err = s.decodeShardIngestOperations(ops)
if err != nil {
return nil, err
}
}
return out, nil
}
func (s Serializer) decodeShardIngestOperations(ops *pb.ShardIngestOperations) ([]*ingest.Operation, error) {
out := []*ingest.Operation{}
if len(ops.Ops) == 0 {
return out, nil
}
for _, op := range ops.Ops {
if op == nil {
continue
}
decoded, err := s.decodeShardIngestOperation(op)
if err != nil {
return nil, err
}
out = append(out, decoded)
}
return out, nil
}
func (s Serializer) decodeShardIngestOperation(op *pb.ShardIngestOperation) (*ingest.Operation, error) {
opType, err := ingest.ParseOpType(op.OpType)
if err != nil {
return nil, err
}
out := &ingest.Operation{
OpType: opType,
ClearRecordIDs: op.ClearRecordIDs,
ClearFields: op.ClearFields,
FieldOps: make(map[string]*ingest.FieldOperation, len(op.FieldOps)),
}
for k, v := range op.FieldOps {
out.FieldOps[k] = &ingest.FieldOperation{
RecordIDs: v.RecordIDs,
Values: v.Values,
Signed: v.Signed,
}
}
return out, nil
}

View file

@ -0,0 +1,149 @@
// Copyright 2021 Molecula Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package proto
import (
"errors"
"reflect"
"testing"
"github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/ingest"
)
func testOneRoundTrip(t *testing.T, s pilosa.Serializer, obj pilosa.Message, expectedMarshalErr error, expectedUnmarshalErr error, expectedMismatchErr error) {
repr, err := s.Marshal(obj)
if err != nil {
if expectedMarshalErr == nil {
t.Fatalf("unexpected marshalling error %q", err.Error())
}
if err.Error() != expectedMarshalErr.Error() {
t.Fatalf("expecting marshalling error %q, got %q", expectedMarshalErr.Error(), err.Error())
}
} else {
if expectedMarshalErr != nil {
t.Fatalf("expected marshalling error %q, got no error", expectedMarshalErr.Error())
}
}
obj2 := reflect.New(reflect.TypeOf(obj).Elem()).Interface()
err = s.Unmarshal(repr, obj2)
if err != nil {
if expectedUnmarshalErr == nil {
t.Fatalf("unexpected unmarshalling error %q", err.Error())
}
if err.Error() != expectedUnmarshalErr.Error() {
t.Fatalf("expecting unmarshalling error %q, got %q", expectedUnmarshalErr.Error(), err.Error())
}
} else {
if expectedUnmarshalErr != nil {
t.Fatalf("expected unmarshalling error %q, got no error", expectedUnmarshalErr.Error())
}
}
switch real := obj.(type) {
case *ingest.ShardedRequest:
real2 := obj2.(*ingest.ShardedRequest)
err := real.Compare(real2)
if err != nil {
if expectedMismatchErr == nil {
t.Fatalf("unexpected compare error %q", err.Error())
}
if err.Error() != expectedMismatchErr.Error() {
t.Fatalf("expecting compare error %q, got %q", expectedMismatchErr.Error(), err.Error())
}
} else {
if expectedMismatchErr != nil {
t.Fatalf("expected compare error %q, got no error", expectedMismatchErr.Error())
}
}
default:
if !reflect.DeepEqual(obj, obj2) {
t.Fatalf("serialization round trip failed for %T:\nexpected %#v\ngot %#v", obj, obj, obj2)
}
}
}
type shardedIngestRequestTest struct {
req *ingest.ShardedRequest
err error
}
var shardedIngestRequestTestcases = []shardedIngestRequestTest{
{
req: &ingest.ShardedRequest{
Ops: map[uint64][]*ingest.Operation{
1: {
{
OpType: ingest.OpWrite,
ClearFields: []string{"clearField", "clearField2"},
ClearRecordIDs: []uint64{1, 7, 9},
FieldOps: map[string]*ingest.FieldOperation{
"writeAll": {
RecordIDs: []uint64{3, 6, 8},
Values: []uint64{0, 17, 34},
Signed: []int64{-9, 23, 17},
},
"writeValues": {
RecordIDs: []uint64{3, 6, 8},
Values: []uint64{0, 17, 34},
},
"writeSigned": {
RecordIDs: []uint64{3, 6, 8},
Signed: []int64{-9, 23, 17},
},
},
},
{
OpType: ingest.OpSet,
FieldOps: map[string]*ingest.FieldOperation{
"foo": nil,
},
},
},
2: {},
3: nil,
},
},
},
{
req: &ingest.ShardedRequest{
Ops: map[uint64][]*ingest.Operation{
1: {
{
OpType: ingest.OpWrite,
FieldOps: map[string]*ingest.FieldOperation{
"writeAll": {},
},
},
{
OpType: ingest.OpSet,
FieldOps: map[string]*ingest.FieldOperation{
"foo": nil,
},
},
nil,
},
},
},
err: errors.New("shard 1: expected 3 ops, got 2"),
},
}
func TestIngestRoundTrip(t *testing.T) {
for _, tc := range shardedIngestRequestTestcases {
t.Logf("next case")
testOneRoundTrip(t, DefaultSerializer, tc.req, nil, nil, tc.err)
}
}

View file

@ -32,6 +32,7 @@ import (
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/encoding/proto"
"github.com/molecula/featurebase/v2/ingest"
pnet "github.com/molecula/featurebase/v2/net"
"github.com/molecula/featurebase/v2/topology"
"github.com/molecula/featurebase/v2/tracing"
@ -284,6 +285,38 @@ func (c *InternalClient) IngestOperations(ctx context.Context, uri *pnet.URI, in
return nil
}
// IngestNodeOperations uses the internal/protobuf ingest endpoint for ingest data
func (c *InternalClient) IngestNodeOperations(ctx context.Context, uri *pnet.URI, indexName string, ireq *ingest.ShardedRequest) error {
if uri == nil {
uri = c.defaultURI
}
u := uri.Path(fmt.Sprintf("/internal/ingest/%s/node", indexName))
buf, err := c.serializer.Marshal(ireq)
if err != nil {
return errors.Wrap(err, "marshalling")
}
req, err := http.NewRequest("POST", u, bytes.NewReader(buf))
if err != nil {
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return errors.Errorf("unexpected status code: %s", resp.Status)
}
return nil
}
// MutexCheck uses the mutex-check endpoint to request mutex collision data
// from a single node. It produces per-shard results, and does not translate
// them.

View file

@ -42,6 +42,7 @@ import (
"github.com/gorilla/mux"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/encoding/proto"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/logger"
"github.com/molecula/featurebase/v2/pql"
"github.com/molecula/featurebase/v2/rbf"
@ -432,7 +433,9 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/internal/index/{index}/shards", handler.handleGetIndexAvailableShards).Methods("GET").Name("GetIndexAvailableShards")
router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes")
router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client
router.HandleFunc("/internal/ingest/{index}", handler.handleIngestData).Methods("POST").Name("PostIngestData")
router.HandleFunc("/internal/ingest/{index}", handler.handlePostIngestData).Methods("POST").Name("PostIngestData")
router.HandleFunc("/internal/ingest/{index}/node", handler.handlePostIngestNode).Methods("POST").Name("PostIngestNode")
router.HandleFunc("/internal/schema", handler.handleIngestSchema).Methods("POST").Name("PostIngestSchema")
router.HandleFunc("/internal/translate/index/{index}/keys/find", handler.handleFindIndexKeys).Methods("POST").Name("FindIndexKeys")
router.HandleFunc("/internal/translate/index/{index}/keys/create", handler.handleCreateIndexKeys).Methods("POST").Name("CreateIndexKeys")
@ -1337,7 +1340,9 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) {
resp.write(w, err)
}
func (h *Handler) handleIngestData(w http.ResponseWriter, r *http.Request) {
// handlePostIngestData handles JSON ingest data that may need key
// translation, for the entire cluster.
func (h *Handler) handlePostIngestData(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
@ -1351,6 +1356,15 @@ func (h *Handler) handleIngestData(w http.ResponseWriter, r *http.Request) {
qcx := h.api.Txf().NewQcx()
err := h.api.IngestOperations(r.Context(), qcx, indexName, r.Body)
if err == nil {
err = qcx.Finish()
if err != nil {
http.Error(w, fmt.Sprintf("ingesting: %v", err), http.StatusInternalServerError)
return
}
} else {
qcx.Abort()
}
resp := successResponse{h: h, Name: indexName}
resp.write(w, err)
@ -2908,6 +2922,52 @@ func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request
}
}
// handlePostIngestNode is the internal endpoint taking already-translated
// ingest operations, sorted by shard, for a single node.
func (h *Handler) handlePostIngestNode(w http.ResponseWriter, r *http.Request) {
// Verify that request is only communicating over protobufs.
if error, code := validateProtobufHeader(r); error != "" {
http.Error(w, error, code)
return
}
ctx := r.Context()
// Read entire body.
span, _ := tracing.StartSpanFromContext(ctx, "ioutil.ReadAll-Body")
body, err := readBody(r)
span.LogKV("bodySize", len(body))
span.Finish()
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
req := &ingest.ShardedRequest{}
span, _ = tracing.StartSpanFromContext(ctx, "Unmarshal")
err = proto.DefaultSerializer.Unmarshal(body, req)
span.Finish()
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
urlVars := mux.Vars(r)
indexName := urlVars["index"]
qcx := h.api.Txf().NewQcx()
err = h.api.IngestNodeOperations(r.Context(), qcx, indexName, req)
if err == nil {
err = qcx.Finish()
if err != nil {
http.Error(w, fmt.Sprintf("ingesting: %v", err), http.StatusInternalServerError)
}
} else {
http.Error(w, fmt.Sprintf("ingesting: %v", err), http.StatusInternalServerError)
qcx.Abort()
}
}
func (h *Handler) handlePostTranslateKeys(w http.ResponseWriter, r *http.Request) {
// Verify that request is only communicating over protobufs.
if r.Header.Get("Content-Type") != "application/x-protobuf" {

View file

@ -15,10 +15,13 @@
package ingest
import (
"bytes"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"math"
"sort"
"strconv"
"time"
@ -37,15 +40,22 @@ import (
// decimal yes yes no
// timestamp yes yes no
type KeyLookupFunc func(...string) (map[string]uint64, error)
// KeyTranslator is a thing that can translate strings to IDs, and also
// IDs back to strings. The ID->string conversion is used only to render
// a request back to JSON, which in turn is only used in testing. The
// functions optionally take an existing map which they then augment.
type KeyTranslator interface {
TranslateKeys(keys ...string) (map[string]uint64, error)
TranslateIDs(ids ...uint64) (map[uint64]string, error)
}
// Codec is a single-use parser which decodes data into columnar vectors.
type Codec interface {
AddSetField(name string, lookup KeyLookupFunc) error
AddTimeQuantumField(name string, lookup KeyLookupFunc) error
AddMutexField(name string, lookup KeyLookupFunc) error
AddSetField(name string, keys KeyTranslator) error
AddTimeQuantumField(name string, keys KeyTranslator) error
AddMutexField(name string, keys KeyTranslator) error
AddBoolField(name string) error
AddIntField(name string, lookup KeyLookupFunc) error
AddIntField(name string, keys KeyTranslator) error
AddDecimalField(name string, scale int64) error
AddTimestampField(name string, scale time.Duration, epoch int64) error
@ -56,138 +66,299 @@ type Codec interface {
type jsonDecFn func(recID uint64, typ jsonparser.ValueType, data []byte) error
// applyTranslationFn is a function which applies key lookups to the values
// of an operation, meaning it needs to know whether it's applying them
// to the signed or unsigned values.
type applyTranslationFn func(*FieldOperation, []uint64) error
// jsonEncFn encodes a value, or range of values, according to a given
// fieldCodec's translation rules. key values, if present, will be used to
// replace whichever values could have been provided as keys. so for instance,
// with an int field which is keyed, values are actually of type int64, but
// if strings are present, those are used. for a time quantum field, the time
// is set from the signed field, and the value is replaced by the key if
// keys are provided.
type jsonEncFn func(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error
type jsonFieldCodec struct {
// fieldCodec represents something that can encode and decode a
// particular field. Its methods are not reentrant, it uses internal
// buffers.
type fieldCodec struct {
valueKeys *StringTable
currentOp *FieldOperation
decode jsonDecFn
translate applyTranslationFn
encode jsonEncFn
// For timestamp: scale-in-nanoseconds; for instance, if scaleUnit is
// 1,000,000,000, we are storing numbers-of-seconds since the Unix epoch.
// The actual value recorded in BSI will be offset by the field's
// epoch, but we don't need to know that.
// For decimal: Decimal digits of precision. So for instance, with
// scaleUnit 2, "1" is stored as 100 and "1.2" is stored as 120.
// scale 2, scaleUnit is 100, "1" is stored as 100 and "1.2" is stored as
// 120.
scaleUnit int64
scale int64
epoch int64 // used only by Timestamp fields
scratch []uint64 // reusable scratch space for sets of values
lookup KeyLookupFunc
keys KeyTranslator
buf []byte // scratch space for format operations.
fieldType FieldType
}
// JSONCodec is a Codec which accepts a JSON map of record keys/ids to updated value maps.
type JSONCodec struct {
recKeys *StringTable
fields map[string]*jsonFieldCodec
keyLookup KeyLookupFunc
currentOp *Operation
fieldTypes map[string]FieldType
recKeys *StringTable
fields map[string]*fieldCodec
keys KeyTranslator
currentOp *Operation
}
// jsonBuffer is a bytes.Buffer which has an associated json.Encoder which
// can be used to write to its buffer. Both types are just embedded because
// they have non-overlapping APIs. Please don't look at my horrible face.
//
// This has some internal objects it can use to stash things
// so that it can pass &foo to enc.Encode() without needing to heap-allocate
// something for the interface conversion, and a buffer it can use for
// converting numbers or times.
//
// It also has an internal error buffer. In fact, in many cases, errors
// are so far as I can tell absolutely impossible -- bytes.Buffer specifically
// promises never to yield an error, and nothing seems to hint that
// enc.Encode can error on integers, strings, booleans, or arrays. So we
// have dozens of error checks that we can't cause to happen even with
// malformed inputs... so we eat those errors, don't require them to be
// checked externally, and return them when done if anyone checks.
type jsonBuffer struct {
*bytes.Buffer
enc *json.Encoder
buf [48]byte // scratch space for using strconv.AppendInt, etc
uintbuf []uint64 // dummy buffer so that we don't have to alloc to print []uint64
strbuf []string // dummy buffer so that we don't have to alloc to print []string
str string // and again, "so we don't have to alloc a copy"
err error
}
// Encode removes the stray newlines added by json.Encoder.
func (j *jsonBuffer) Encode(v interface{}) {
err := j.enc.Encode(v)
if err == nil {
j.Truncate(j.Len() - 1)
} else {
j.err = err
}
}
// EncodeInt uses AppendInt into a static buffer to reduce allocs.
func (j *jsonBuffer) EncodeInt(i int64) {
rep := strconv.AppendInt(j.buf[:0], i, 10)
_, _ = j.Write(rep)
}
// EncodeUint uses AppendUint into a static buffer to reduce allocs.
func (j *jsonBuffer) EncodeUint(u uint64) {
rep := strconv.AppendUint(j.buf[:0], u, 10)
_, _ = j.Write(rep)
}
// EncodeUints uses a static copy of a []uint64 -- the slice, not its
// contents -- in already-allocated memory so the interface conversion
// doesn't have to do that.
func (j *jsonBuffer) EncodeUints(u []uint64) {
j.uintbuf = u
j.Encode(&j.uintbuf)
j.strbuf = nil
}
// EncodeStrings uses a static copy of a []string -- the slice, not its
// contents -- in already-allocated memory so the interface conversion
// doesn't have to do that.
func (j *jsonBuffer) EncodeStrings(s []string) {
j.strbuf = s
j.Encode(&j.strbuf)
j.strbuf = nil
}
// EncodeQuotedUint uses AppendUint into a static buffer to reduce allocs,
// while also surrounding the value with quotes.
func (j *jsonBuffer) EncodeQuotedUint(u uint64) {
j.buf[0] = '"'
rep := strconv.AppendUint(j.buf[1:1], u, 10)
j.buf[len(rep)+1] = '"'
_, _ = j.Write(j.buf[:len(rep)+2])
}
// EncodeString appends the JSON encoding of a string. This exists
// because otherwise runtime allocates a heap-allocated copy of the
// string to live inside an interface{} for the duration of a function
// call...
func (j *jsonBuffer) EncodeString(s string) {
j.str = s
j.Encode(&j.str)
j.str = ""
}
// EncodeTime exists because time's MarshalJSON allocates a new
// buffer every time it gets called, resulting in a full 13% of
// all the allocations produced in a test run, plus another 13%
// or so of them which were for the copies of the time objects
// made to stuff them into an interface{}. Eww.
func (j *jsonBuffer) EncodeTime(t time.Time) {
j.buf[0] = '"'
rep := t.AppendFormat(j.buf[1:1], time.RFC3339Nano)
j.buf[len(rep)+1] = '"'
_, _ = j.Write(j.buf[:len(rep)+2])
}
// EncodeBool writes a literal representation directly to reduce allocs.
func (j *jsonBuffer) EncodeBool(b bool) {
if b {
_, _ = j.WriteString("true")
} else {
_, _ = j.WriteString("false")
}
}
func (j *jsonBuffer) Err() error {
return j.err
}
func newJSONBuffer(data []byte) *jsonBuffer {
e := &jsonBuffer{Buffer: bytes.NewBuffer(data)}
e.enc = json.NewEncoder(e.Buffer)
e.enc.SetEscapeHTML(false) // we are not doing HTML, just JSON
return e
}
var _ Codec = &JSONCodec{}
func NewJSONCodec(lookup KeyLookupFunc) (*JSONCodec, error) {
j := &JSONCodec{fields: map[string]*jsonFieldCodec{}}
if lookup != nil {
func NewJSONCodec(keys KeyTranslator) (*JSONCodec, error) {
j := &JSONCodec{
fields: map[string]*fieldCodec{},
fieldTypes: map[string]FieldType{},
}
if keys != nil {
j.recKeys = NewStringTable()
j.keyLookup = lookup
j.keys = keys
}
return j, nil
}
func (codec *JSONCodec) AddTimeQuantumField(name string, lookup KeyLookupFunc) error {
fieldCodec := &jsonFieldCodec{}
func (codec *JSONCodec) AddTimeQuantumField(name string, keys KeyTranslator) error {
fieldCodec := &fieldCodec{
fieldType: FieldTypeTimeQuantum,
keys: keys,
}
fieldCodec.decode = fieldCodec.DecodeTimeQuantumValue
if lookup != nil {
fieldCodec.translate = (*FieldOperation).TranslateUnsigned
}
return codec.addField(name, FieldTypeTimeQuantum, fieldCodec, lookup)
fieldCodec.encode = fieldCodec.EncodeTimeQuantumValue
return codec.addField(name, fieldCodec)
}
func (codec *JSONCodec) AddSetField(name string, lookup KeyLookupFunc) error {
fieldCodec := &jsonFieldCodec{}
func (codec *JSONCodec) AddSetField(name string, keys KeyTranslator) error {
fieldCodec := &fieldCodec{
fieldType: FieldTypeSet,
keys: keys,
}
fieldCodec.decode = fieldCodec.DecodeSetValue
if lookup != nil {
fieldCodec.translate = (*FieldOperation).TranslateUnsigned
}
return codec.addField(name, FieldTypeSet, fieldCodec, lookup)
fieldCodec.encode = fieldCodec.EncodeSetValue
return codec.addField(name, fieldCodec)
}
func (codec *JSONCodec) AddIntField(name string, lookup KeyLookupFunc) error {
fieldCodec := &jsonFieldCodec{}
func (codec *JSONCodec) AddIntField(name string, keys KeyTranslator) error {
fieldCodec := &fieldCodec{
fieldType: FieldTypeInt,
keys: keys,
}
fieldCodec.decode = fieldCodec.DecodeIntValue
if lookup != nil {
fieldCodec.translate = (*FieldOperation).TranslateSigned
}
return codec.addField(name, FieldTypeInt, fieldCodec, lookup)
fieldCodec.encode = fieldCodec.EncodeIntValue
return codec.addField(name, fieldCodec)
}
func (codec *JSONCodec) AddMutexField(name string, lookup KeyLookupFunc) error {
fieldCodec := &jsonFieldCodec{}
fieldCodec.decode = fieldCodec.DecodeMutexValue
if lookup != nil {
fieldCodec.translate = (*FieldOperation).TranslateUnsigned
func (codec *JSONCodec) AddMutexField(name string, keys KeyTranslator) error {
fieldCodec := &fieldCodec{
fieldType: FieldTypeMutex,
keys: keys,
}
return codec.addField(name, FieldTypeMutex, fieldCodec, lookup)
fieldCodec.decode = fieldCodec.DecodeMutexValue
fieldCodec.encode = fieldCodec.EncodeMutexValue
return codec.addField(name, fieldCodec)
}
func (codec *JSONCodec) AddBoolField(name string) error {
fieldCodec := &jsonFieldCodec{}
fieldCodec := &fieldCodec{
fieldType: FieldTypeBool,
}
fieldCodec.decode = fieldCodec.DecodeBoolValue
return codec.addField(name, FieldTypeBool, fieldCodec, nil)
fieldCodec.encode = fieldCodec.EncodeBoolValue
return codec.addField(name, fieldCodec)
}
// TimestampField is used to store seconds since unix epoch. The numeric values
// stored are adjusted based on the given time.Duration; for instance, if the
// time scale is time.Second, then the second after the epoch is stored as 1,
// if it's time.Millisecond, then it's stored as 1000, etcetera.
// if it's time.Millisecond, then it's stored as 1000, etcetera. The epoch
// passed to this function should be the offset from the Unix epoch to the
// desired epoch, in the same scale. (So if the scale is milliseconds,
// it should be the Unix timestamp in seconds, times 1000.)
func (codec *JSONCodec) AddTimestampField(name string, timeScale time.Duration, epoch int64) error {
fieldCodec := &jsonFieldCodec{scaleUnit: int64(timeScale), epoch: epoch}
fieldCodec := &fieldCodec{
fieldType: FieldTypeTimeStamp,
scaleUnit: int64(timeScale),
epoch: epoch,
}
fieldCodec.decode = fieldCodec.DecodeTimeValue
return codec.addField(name, FieldTypeTimeStamp, fieldCodec, nil)
fieldCodec.encode = fieldCodec.EncodeTimeValue
return codec.addField(name, fieldCodec)
}
// AddDecimalField adds a decimal field, which is stored as integer values
// with a scale offset, but parsed as floating point values. For instance,
// with decimalScale=2, `0.01` would store the value 1.
func (codec *JSONCodec) AddDecimalField(name string, decimalScale int64) error {
fieldCodec := &jsonFieldCodec{scaleUnit: int64(math.Pow10(int(decimalScale)))}
fieldCodec := &fieldCodec{
fieldType: FieldTypeDecimal,
scale: decimalScale,
scaleUnit: int64(math.Pow(10, float64(decimalScale))),
}
fieldCodec.decode = fieldCodec.DecodeDecimalValue
return codec.addField(name, FieldTypeDecimal, fieldCodec, nil)
fieldCodec.encode = fieldCodec.EncodeDecimalValue
return codec.addField(name, fieldCodec)
}
func (codec *JSONCodec) addField(name string, fieldType FieldType, fieldCodec *jsonFieldCodec, lookup KeyLookupFunc) error {
func (codec *JSONCodec) addField(name string, fieldCodec *fieldCodec) error {
if _, ok := codec.fields[name]; ok {
return fmt.Errorf("duplicate field %q", name)
}
if lookup != nil {
if fieldCodec.keys != nil {
fieldCodec.valueKeys = NewStringTable()
fieldCodec.lookup = lookup
}
fieldCodec.fieldType = fieldType
codec.fields[name] = fieldCodec
codec.fieldTypes[name] = fieldCodec.fieldType
return nil
}
// decodeSetOrValue decodes a value which might be either an array of values or a
// single value, where values might be string keys or bare numbers, calling cb for
// each value it finds.
func (j *jsonFieldCodec) decodeSetOrValue(dataType jsonparser.ValueType, data []byte, cb func(uint64) error) (err error) {
func (j *fieldCodec) decodeSetOrValue(dataType jsonparser.ValueType, data []byte, cb func(uint64) error) (err error) {
switch dataType {
case jsonparser.Array:
_, arrayErr := jsonparser.ArrayEach(data, func(value []byte, dataType jsonparser.ValueType, offset int, unused error) {
id, idErr := j.valueKeys.ID(value)
if idErr != nil {
err = idErr
return
}
// stash an error if we got one
valueErr := cb(id)
if valueErr != nil {
err = valueErr
var id uint64
switch dataType {
case jsonparser.String:
id, err = j.valueKeys.ID(value)
if err != nil {
return
}
err = cb(id)
case jsonparser.Number:
if j.valueKeys != nil {
err = errors.New("expecting key, got numeric value")
return
}
id, err = strconv.ParseUint(pretendByteIsString(value), 10, 64)
if err != nil {
return
}
err = cb(id)
default:
err = fmt.Errorf("expecting value or array, got %v", dataType)
}
})
if err != nil {
@ -212,23 +383,32 @@ func (j *jsonFieldCodec) decodeSetOrValue(dataType jsonparser.ValueType, data []
}
return cb(id)
default:
return fmt.Errorf("expecting array, got %v", dataType)
return fmt.Errorf("expecting value or array, got %v", dataType)
}
return err
}
// DecodeSetValue decodes a set of unsigned values from the provided data into the
// associated currentOp.
func (j *jsonFieldCodec) DecodeSetValue(recID uint64, dataType jsonparser.ValueType, data []byte) (err error) {
func (j *fieldCodec) DecodeSetValue(recID uint64, dataType jsonparser.ValueType, data []byte) (err error) {
return j.decodeSetOrValue(dataType, data, func(id uint64) error {
j.currentOp.AddPair(recID, id)
return nil
})
}
func (j *fieldCodec) EncodeSetValue(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error {
if len(keys) == 1 || len(values) == 1 {
// simplify: just hand this off to a single-value case
return j.EncodeMutexValue(dst, values, signed, keys)
}
appendKeysJSON(dst, values, keys)
return nil
}
// DecodeIntValue decodes a single signed value from the provided data
// into the associated currentOp.
func (j *jsonFieldCodec) DecodeIntValue(recID uint64, dataType jsonparser.ValueType, data []byte) error {
func (j *fieldCodec) DecodeIntValue(recID uint64, dataType jsonparser.ValueType, data []byte) error {
switch dataType {
case jsonparser.String:
value, err := j.valueKeys.IntID(data)
@ -255,25 +435,62 @@ func (j *jsonFieldCodec) DecodeIntValue(recID uint64, dataType jsonparser.ValueT
return nil
}
func (j *fieldCodec) EncodeIntValue(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error {
if len(keys) == 0 {
if len(signed) == 0 {
return errors.New("encodeIntValue: need a value")
}
dst.EncodeInt(signed[0])
return nil
}
dst.EncodeString(keys[0])
return nil
}
// DecodeMutexValue decodes a single unsigned value from the provided data
// into the associated currentOp.
func (j *jsonFieldCodec) DecodeMutexValue(recID uint64, dataType jsonparser.ValueType, data []byte) error {
func (j *fieldCodec) DecodeMutexValue(recID uint64, dataType jsonparser.ValueType, data []byte) error {
switch dataType {
case jsonparser.Number, jsonparser.String:
id, err := j.valueKeys.ID(data)
case jsonparser.String:
value, err := j.valueKeys.ID(data)
if err != nil {
return err
}
j.currentOp.AddPair(recID, id)
j.currentOp.AddPair(recID, value)
case jsonparser.Number:
if j.valueKeys != nil {
return errors.New("expecting string key, got numeric value")
}
value, err := strconv.ParseUint(pretendByteIsString(data), 10, 64)
if err != nil {
return err
}
j.currentOp.AddPair(recID, value)
default:
return fmt.Errorf("expecting integer value, got %v", dataType)
if j.valueKeys != nil {
return fmt.Errorf("expecting string key, got %v", dataType)
} else {
return fmt.Errorf("expecting integer value, got %v", dataType)
}
}
return nil
}
func (j *fieldCodec) EncodeMutexValue(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error {
if len(keys) == 0 {
if len(values) == 0 {
return errors.New("encodeMutexValue: need a value")
}
dst.EncodeUint(values[0])
return nil
}
dst.EncodeString(keys[0])
return nil
}
// DecodeBoolValue decodes a single true/false value from the provided data
// into the associated currentOp.
func (j *jsonFieldCodec) DecodeBoolValue(recID uint64, typ jsonparser.ValueType, data []byte) error {
func (j *fieldCodec) DecodeBoolValue(recID uint64, typ jsonparser.ValueType, data []byte) error {
value := uint64(0)
switch typ {
case jsonparser.String:
@ -303,9 +520,21 @@ func (j *jsonFieldCodec) DecodeBoolValue(recID uint64, typ jsonparser.ValueType,
return nil
}
func (j *fieldCodec) EncodeBoolValue(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error {
if len(values) == 0 {
return errors.New("encoding boolean value, but none provided")
}
var x bool
if values[0] != 0 {
x = true
}
dst.EncodeBool(x)
return nil
}
// DecodeTimeQuantumValue decodes a timestamp, and a set of bits from the
// provided data into the associated currentOp.
func (j *jsonFieldCodec) DecodeTimeQuantumValue(recID uint64, typ jsonparser.ValueType, data []byte) error {
func (j *fieldCodec) DecodeTimeQuantumValue(recID uint64, typ jsonparser.ValueType, data []byte) error {
j.scratch = j.scratch[:0]
stamp := time.Unix(0, 0).UTC()
err := jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) (err error) {
@ -346,8 +575,27 @@ func (j *jsonFieldCodec) DecodeTimeQuantumValue(recID uint64, typ jsonparser.Val
return nil
}
// Our format does not allow a record to have multiple values set with
// different timestamps at the same time. We use the first timestamp for
// the whole set.
func (j *fieldCodec) EncodeTimeQuantumValue(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error {
if len(values) == 0 {
return errors.New("encoding time quantum value: no value provided")
}
dst.WriteString(`{"values":`)
appendKeysJSON(dst, values, keys)
// a zero timestamp is idiomatic for no-time-provided, and i sort of
// hate that, but here we are.
if len(signed) > 0 && signed[0] != 0 {
_, _ = dst.WriteString(`,"time":`)
dst.EncodeTime(time.Unix(0, signed[0]).UTC())
}
_, _ = dst.WriteString(`}`)
return nil
}
// DecodeTimeValue will eventually work but right now it doesn't actually.
func (j *jsonFieldCodec) DecodeTimeValue(recID uint64, dataType jsonparser.ValueType, data []byte) (err error) {
func (j *fieldCodec) DecodeTimeValue(recID uint64, dataType jsonparser.ValueType, data []byte) (err error) {
var stamp time.Time
switch dataType {
case jsonparser.String:
@ -364,12 +612,22 @@ func (j *jsonFieldCodec) DecodeTimeValue(recID uint64, dataType jsonparser.Value
return fmt.Errorf("parsing numeric timestamp: %w", err)
}
j.currentOp.AddSignedPair(recID, i64)
default:
return fmt.Errorf("expecting time, got %s", dataType)
}
return nil
}
func (j *fieldCodec) EncodeTimeValue(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error {
if len(signed) == 0 {
return errors.New("encoding time value: no value provided")
}
dst.EncodeTime(time.Unix(0, (signed[0]+j.epoch)*j.scaleUnit).UTC())
return nil
}
// DecodeDecimalValue will eventually work but right now it doesn't actually.
func (j *jsonFieldCodec) DecodeDecimalValue(recID uint64, dataType jsonparser.ValueType, data []byte) error {
func (j *fieldCodec) DecodeDecimalValue(recID uint64, dataType jsonparser.ValueType, data []byte) error {
switch dataType {
case jsonparser.String, jsonparser.Number:
value, err := jsonparser.GetFloat(data)
@ -384,6 +642,28 @@ func (j *jsonFieldCodec) DecodeDecimalValue(recID uint64, dataType jsonparser.Va
return nil
}
func (j *fieldCodec) EncodeDecimalValue(dst *jsonBuffer, values []uint64, signed []int64, keys []string) error {
j.buf = strconv.AppendInt(j.buf[:0], signed[0], 10)
scale := int(j.scale)
if len(j.buf) > scale {
j.buf = append(j.buf, '.')
// shove the last scaleUnit values over
copy(j.buf[len(j.buf)-scale:], j.buf[len(j.buf)-scale-1:])
j.buf[len(j.buf)-scale-1] = '.'
}
_, _ = dst.Write(j.buf)
return nil
}
// FieldTypes gives a mapping of fields to their basic types used by this
// codec.
func (codec *JSONCodec) FieldTypes() map[string]FieldType {
return codec.fieldTypes
}
// ParseKeyedRecords parses the records it finds. Note that, when you're doing
// a Write op, this will update ClearRecordIDs automatically as it goes,
// even though record_ids isn't specified in the JSON for that case.
func (codec *JSONCodec) ParseKeyedRecords(data []byte) (err error) {
seen := make(map[uint64]struct{})
return jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error {
@ -412,8 +692,8 @@ func (codec *JSONCodec) ParseKeyedRecords(data []byte) (err error) {
})
}
func (codec *JSONCodec) ParseOperation(data []byte) (op *Operation, err error) {
op = &Operation{FieldOps: make(map[string]*FieldOperation, len(codec.fields))}
func (codec *JSONCodec) ParseOperation(data []byte, seq int) (op *Operation, err error) {
op = &Operation{FieldOps: make(map[string]*FieldOperation, len(codec.fields)), Seq: seq}
codec.currentOp = op
err = jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error {
switch string(key) {
@ -491,10 +771,12 @@ func (codec *JSONCodec) Parse(r io.Reader) (req *Request, err error) {
func (codec *JSONCodec) ParseBytes(data []byte) (req *Request, err error) {
var ops []*Operation
var lastErr error
var seq int
_, err = jsonparser.ArrayEach(data, func(value []byte, dataType jsonparser.ValueType, offset int, err error) {
switch dataType {
case jsonparser.Object:
op, err := codec.ParseOperation(value)
op, err := codec.ParseOperation(value, seq)
seq++
if err != nil {
lastErr = fmt.Errorf("parsing operation: %v", err)
return
@ -512,34 +794,39 @@ func (codec *JSONCodec) ParseBytes(data []byte) (req *Request, err error) {
}
// and now, key translation!
var keyMap []uint64
if codec.keyLookup != nil {
keyMap, err = MapForStringTable(codec.recKeys, codec.keyLookup)
if codec.keys != nil {
keyMap, err = codec.recKeys.MakeIDMap(codec.keys)
if err != nil {
return nil, fmt.Errorf("trying to find record key mapping: %w", err)
}
}
req = &Request{FieldTypes: make(map[string]FieldType, len(codec.fields))}
req = &Request{}
valueMaps := map[string]func(*FieldOperation) error{}
for name, fieldCodec := range codec.fields {
// make closure survive iteration
fieldCodec := fieldCodec
if fieldCodec.lookup != nil {
fieldMap, err := MapForStringTable(fieldCodec.valueKeys, fieldCodec.lookup)
if fieldCodec.keys != nil {
fieldMap, err := fieldCodec.valueKeys.MakeIDMap(fieldCodec.keys)
if err != nil {
return nil, fmt.Errorf("trying to find value mapping for %q: %w", name, err)
}
valueMaps[name] = func(fo *FieldOperation) error {
return fieldCodec.translate(fo, fieldMap)
if fieldCodec.fieldType == FieldTypeInt {
valueMaps[name] = func(fo *FieldOperation) error {
return translateSigned(fieldMap, fo.Signed)
}
} else {
valueMaps[name] = func(fo *FieldOperation) error {
return translateUnsigned(fieldMap, fo.Values)
}
}
}
req.FieldTypes[name] = fieldCodec.fieldType
}
for _, op := range ops {
// For Clear and Write, we need to translate/sort our record ID
// list.
if op.OpType == OpClear || op.OpType == OpWrite || op.OpType == OpDelete {
if keyMap != nil {
if err = translateUnsignedSlice(op.ClearRecordIDs, keyMap); err != nil {
if err = translateUnsigned(keyMap, op.ClearRecordIDs); err != nil {
return nil, fmt.Errorf("mapping record keys for clear op: %w", err)
}
}
@ -554,7 +841,7 @@ func (codec *JSONCodec) ParseBytes(data []byte) (req *Request, err error) {
continue
}
if keyMap != nil {
if err = fieldOp.TranslateKeys(keyMap); err != nil {
if err = translateUnsigned(keyMap, fieldOp.RecordIDs); err != nil {
return nil, fmt.Errorf("mapping record keys for op on %q: %w", field, err)
}
}
@ -574,6 +861,283 @@ func (codec *JSONCodec) ParseBytes(data []byte) (req *Request, err error) {
return req, nil
}
// AppendBytes appends bytes which we would expect to produce the same
// request, using this codec. It does not try to recreate FieldTypes, which
// would be handled by the codec anyway. It's written as an Append so you
// can reuse a buffer.
func (codec *JSONCodec) AppendBytes(req *Request, data []byte) (out []byte, err error) {
if req == nil || len(req.Ops) == 0 {
return append(data, "[]"...), nil
}
dst := newJSONBuffer(data)
// we ignore the FieldTypes part of the Request, which is just there to
// let the request's ByShard use the correct sorting routines, which is
// itself sort of awful. The codec will insert it again when parsing the
// bytes.
_, _ = dst.WriteString(`,`)
for _, op := range req.Ops {
// EncodeJSON needs to have access to this codec's field data
// and key translators.
err = op.EncodeJSON(dst, codec)
if err != nil {
return nil, err
}
_, _ = dst.WriteString(`,`)
}
// delete the last comma
dst.Truncate(dst.Len() - 1)
_, _ = dst.WriteString(`]`)
// there's very few ways an error could occur, possibly none, but
// just in case lots of internal operations stashed an error if one
// happened, so we'll return anything they came up with.
return dst.Bytes(), dst.Err()
}
// appendIDsJSON appends the provided IDs to a JSON stream as a bracketed
// list of quoted strings if there's a key translator, or integer values
// if the KeyTranslator is nil, comma-separated. It can error because a
// translator can error.
func (codec *JSONCodec) appendIDsJSON(dst *jsonBuffer, values []uint64, keys KeyTranslator) error {
return appendIDsJSON(dst, values, keys)
}
func appendKeysJSON(dst *jsonBuffer, values []uint64, keys []string) {
if len(values) == 0 && len(keys) == 0 {
_, _ = dst.WriteString("[]")
return
}
if len(keys) != 0 {
dst.EncodeStrings(keys)
} else {
dst.EncodeUints(values)
}
}
func appendIDsJSON(dst *jsonBuffer, values []uint64, keys KeyTranslator) error {
if len(values) == 0 {
_, _ = dst.WriteString("[]")
return nil
}
if keys == nil {
dst.EncodeUints(values)
return nil
}
_, _ = dst.WriteString(`[`)
translated, err := keys.TranslateIDs(values...)
if err != nil {
return err
}
for _, v := range values {
dst.EncodeString(translated[v])
_, _ = dst.WriteString(`,`)
}
dst.Truncate(dst.Len() - 1)
_, _ = dst.WriteString(`]`)
return nil
}
// RequestByShard makes up for the fact that we don't want to stash the
// field type data in the request, but we need it to actually do the by-shard.
func (codec *JSONCodec) RequestByShard(req *Request) (*ShardedRequest, error) {
return req.ByShard(codec.fieldTypes)
}
// EncodeJSON encodes an operation as JSON using a provided buffer.
// It uses the provided codec where necessary to help with key
// translation.
func (o *Operation) EncodeJSON(dst *jsonBuffer, codec *JSONCodec) (err error) {
// We're not trying to guarantee that what we produce makes sense or is
// identical to what produced us, just to write our current state out.
if o == nil {
_, _ = dst.WriteString(`{}`)
return nil
}
_, _ = dst.WriteString(`{"action":`)
dst.EncodeString(o.OpType.String())
// it is intentional that o.Seq isn't encoded here; it makes no sense
// to allow an op to specify its seq in JSON.
if o.OpType != OpWrite {
// for Write, ClearRecordIDs and ClearFields were computed
// from the records being written, and aren't actually part of the data.
if len(o.ClearRecordIDs) != 0 {
_, _ = dst.WriteString(`,"record_ids":`)
err = codec.appendIDsJSON(dst, o.ClearRecordIDs, codec.keys)
if err != nil {
return err
}
}
if len(o.ClearFields) != 0 {
_, _ = dst.WriteString(`,"fields":`)
dst.EncodeStrings(o.ClearFields)
}
}
if len(o.FieldOps) == 0 {
_, _ = dst.WriteString(`}`)
return
}
// collect all the fields that have a non-empty set of records. we can
// just ignore the others.
fieldNames := make([]string, 0, len(o.FieldOps))
for field, op := range o.FieldOps {
if op != nil && len(op.RecordIDs) != 0 {
fieldNames = append(fieldNames, field)
}
}
// nevermind then
if len(fieldNames) == 0 {
_, _ = dst.WriteString(`}`)
return nil
}
_, _ = dst.WriteString(`,"records":{`)
// now we have to invert the logic, creating records from
// fields with corresponding ops. uh-oh.
fieldOps := make([]*FieldOperation, len(o.FieldOps))
fieldCodecs := make([]*fieldCodec, len(o.FieldOps))
fieldKeys := make([][]string, len(o.FieldOps)) // translated field keys
indexes := make([]int, len(o.FieldOps))
sort.Slice(fieldNames, func(i, j int) bool { return fieldNames[i] < fieldNames[j] })
next := ^uint64(0)
var idKeys map[uint64]string
if codec.keys != nil {
idKeys = make(map[uint64]string)
}
for i, field := range fieldNames {
// populate a parallel slice of field ops so we don't have
// to do map lookups for every single piece of data
op := o.FieldOps[field]
fieldOps[i] = op
fc := codec.fields[field]
if fc == nil {
return fmt.Errorf("unknown field: %q", field)
}
fieldCodecs[i] = fc
if codec.keys != nil {
// we'll build a list of keys we need for any records
for _, v := range op.RecordIDs {
idKeys[v] = ""
}
}
id := op.RecordIDs[0]
if id < next {
next = id
}
if fc.keys != nil {
thisFieldKeys := make([]string, len(op.RecordIDs))
if fc.fieldType == FieldTypeInt {
u := make([]uint64, len(op.Signed))
for i := range op.Signed {
u[i] = uint64(op.Signed[i])
}
valueKeys, err := fc.keys.TranslateIDs(u...)
if err != nil {
return err
}
for j, v := range op.Signed {
thisFieldKeys[j] = valueKeys[uint64(v)]
}
} else {
valueKeys, err := fc.keys.TranslateIDs(op.Values...)
if err != nil {
return err
}
for j, v := range op.Values {
thisFieldKeys[j] = valueKeys[v]
}
}
fieldKeys[i] = thisFieldKeys
}
}
// ... no fieldops actually have any entries. therefore no record will
// have any fields set, therefore no records exist...
if next == ^uint64(0) {
_, _ = dst.WriteString(`}}`)
return nil
}
if codec.keys != nil {
idList := make([]uint64, 0, len(idKeys))
for k := range idKeys {
idList = append(idList, k)
}
idKeys, err = codec.keys.TranslateIDs(idList...)
if err != nil {
return err
}
}
// next is the ID of a record which exists for at least one field.
// we'll recompute it every loop.
for next != ^uint64(0) {
if codec.keys != nil {
dst.EncodeString(idKeys[next])
} else {
dst.EncodeQuotedUint(next)
}
_, _ = dst.WriteString(`:{`)
current := next
next = ^uint64(0)
for i, field := range fieldNames {
op := fieldOps[i]
idx := indexes[i]
if idx >= len(op.RecordIDs) {
continue
}
id := op.RecordIDs[idx]
if id == current {
var j int
// count ahead to first index which is either outside the list
// or a different id
for j = idx; j < len(op.RecordIDs) && op.RecordIDs[j] == id; j++ {
}
// fmt.Printf("field %s encoding %d-%d (v %d, s %d, k %d)\n",
// field, idx, j, len(op.Values), len(op.Signed), len(fieldKeys[i]))
// print this one, and advance this index to next position
dst.EncodeString(field)
_, _ = dst.WriteString(`:`)
var values []uint64
var signed []int64
var keys []string
if len(op.Values) >= j {
values = op.Values[idx:j]
}
if len(op.Signed) >= j {
signed = op.Signed[idx:j]
}
if len(fieldKeys[i]) >= j {
keys = fieldKeys[i][idx:j]
}
err = fieldCodecs[i].encode(dst, values, signed, keys)
if err != nil {
return err
}
_, _ = dst.WriteString(`,`)
indexes[i] = j
// and we'll fall through to the id < next check, so we
// just set id here.
if indexes[i] < len(op.RecordIDs) {
id = op.RecordIDs[indexes[i]]
} else {
id = ^uint64(0)
}
}
if id < next {
next = id
}
}
// we should always have a trailing comma after any entry, and
// if there were no entries we shouldn't have had a list...
dst.Truncate(dst.Len() - 1)
_, _ = dst.WriteString(`},`)
}
dst.Truncate(dst.Len() - 1)
// close both the records list, and the whole object.
_, _ = dst.WriteString(`}}`)
// In theory, there's no way for most of these to produce errors,
// but just in case, we'll check for an error every operation or so.
return dst.Err()
}
type errFieldNotFound struct {
field string
}

File diff suppressed because it is too large Load diff

View file

@ -15,6 +15,7 @@
package ingest
import (
"errors"
"fmt"
"math/bits"
"sort"
@ -92,14 +93,22 @@ func ParseOpType(s string) (OpType, error) {
// record IDs and fields, a Set or Remove will have only FieldOps,
// and a Write will have both -- populating the record IDs and fields
// from the fieldops.
//
// When we parse an operation, we assign each op a sequential ID
// within the overall request. We keep these IDs associated with ops
// when splitting them up across shards, so we can reverse this even
// if some shards don't get some ops.
type Operation struct {
OpType OpType
Seq int // sequence position within a chain of ops
ClearRecordIDs []uint64
ClearFields []string
FieldOps map[string]*FieldOperation
}
// Compare reports whether two operations seem to be the same.
// Compare reports whether two Operations seem to be the same. While
// a FieldOperation can be "empty" and compare-equal-to a nil FieldOperation,
// no Operation is considered empty even if it has no FieldOps.
func (got *Operation) Compare(expected *Operation) error {
if got == nil && expected == nil {
return nil
@ -122,27 +131,39 @@ func (got *Operation) Compare(expected *Operation) error {
return fmt.Errorf("clear record id %d differs: expected %d, got %d", i, v2, v1)
}
}
if len(got.ClearFields) != len(expected.ClearFields) {
return fmt.Errorf("clear field counts differ: expected %d (%q), got %d (%q)", len(expected.ClearFields), expected.ClearFields, len(got.ClearFields), got.ClearFields)
// don't assume consistent ordering for the fields, because they're
// coming out in arbitrary hash order
seenFields := make(map[string]struct{}, len(expected.ClearFields))
for _, v1 := range expected.ClearFields {
seenFields[v1] = struct{}{}
}
for i, v1 := range got.ClearFields {
v2 := expected.ClearFields[i]
if v1 != v2 {
return fmt.Errorf("clear field %d differs: expected %q, got %q", i, v2, v1)
for _, v2 := range got.ClearFields {
if _, ok := seenFields[v2]; !ok {
return fmt.Errorf("field %q cleared unexpectedly", v2)
}
delete(seenFields, v2)
}
for v1 := range seenFields {
return fmt.Errorf("field %q should be cleared but wasn't", v1)
}
// We check compare even if the op we find on one side is nil, so
// a round-trip test will consider a missing op and an empty op to
// be interchangeable.
for k, fo1 := range got.FieldOps {
fo2 := expected.FieldOps[k]
if err := fo1.Compare(fo2); err != nil {
return fmt.Errorf("field %q mismatch: %w", k, err)
}
}
for k := range expected.FieldOps {
_, ok := got.FieldOps[k]
if !ok {
return fmt.Errorf("expected op for field %q, but none found", k)
for k, fo2 := range expected.FieldOps {
fo1 := got.FieldOps[k]
if err := fo1.Compare(fo2); err != nil {
return fmt.Errorf("field %q mismatch: %w", k, err)
}
}
if expected.Seq != got.Seq {
return fmt.Errorf("sequence mismatch: expected %d, got %d", expected.Seq, got.Seq)
}
return nil
}
@ -172,6 +193,40 @@ func (o *Operation) Sort() {
sort.Strings(o.ClearFields)
}
// clone makes a duplicate of the operation without shared storage
func (o *Operation) clone() *Operation {
o2 := &Operation{
OpType: o.OpType,
Seq: o.Seq,
ClearRecordIDs: append([]uint64{}, o.ClearRecordIDs...),
ClearFields: append([]string{}, o.ClearFields...),
FieldOps: make(map[string]*FieldOperation, len(o.FieldOps)),
}
for k, v := range o.FieldOps {
o2.FieldOps[k] = v.clone()
}
return o2
}
// merge merges the fieldops of the provided operation into this operation.
func (o *Operation) merge(o2 *Operation) {
o.ClearRecordIDs = append(o.ClearRecordIDs, o2.ClearRecordIDs...)
o.ClearFields = append(o.ClearFields, o2.ClearFields...)
for f, op := range o2.FieldOps {
dst := o.FieldOps[f]
if dst == nil {
o.FieldOps[f] = op
continue
}
// otherwise append any contents. dst is a pointer-to, so this
// updates the thing the map points to, we don't have to write back
// into the map.
dst.RecordIDs = append(dst.RecordIDs, op.RecordIDs...)
dst.Values = append(dst.Values, op.Values...)
dst.Signed = append(dst.Signed, op.Signed...)
}
}
type ShardedFieldOperation map[uint64]*FieldOperation
// ByShard() divides the FieldOperation's values up into corresponding chunks
@ -184,22 +239,14 @@ func (f *FieldOperation) ByShard() ShardedFieldOperation {
return f.SortToShards()
}
// ShardInto puts the shards it finds into a target map.
func (f *FieldOperation) ShardInto(target ShardedFieldOperation) {
shards, ends := shardwidth.FindShards(f.RecordIDs)
prev := 0
for i, shard := range shards {
endIndex := ends[i]
subOp := &FieldOperation{RecordIDs: f.RecordIDs[prev:endIndex]}
if len(f.Values) > 0 {
subOp.Values = f.Values[prev:endIndex]
}
if len(f.Signed) > 0 {
subOp.Signed = f.Signed[prev:endIndex]
}
target[shard] = subOp
prev = endIndex
// clone makes a duplicate of the operation without shared storage
func (f *FieldOperation) clone() *FieldOperation {
f2 := &FieldOperation{
RecordIDs: append([]uint64{}, f.RecordIDs...),
Values: append([]uint64{}, f.Values...),
Signed: append([]int64{}, f.Signed...),
}
return f2
}
func ShardIDs(ids []uint64) (out map[uint64][]uint64) {
@ -349,10 +396,22 @@ func (f *FieldOperation) SortByKeys(keys []uint64) {
// doesn't help as much as you'd hope. This beats using stdlib sort by about
// a factor of two for those small N, for larger N we're using the radix sort
// that calls this.
//
// External keys exist only when we are sorting by value, which is to say,
// when we're using row-oriented formats (set, mutex, time quantum).
// For int/decimal/timestamp fields, we're sorting by record only.
// So, if keys is the same as f.RecordIDs, we're looking at an int field
// or equivalent, so Signed exists and Values doesn't exist.
// Otherwise, we might be looking at a time quantum field (both exist)
// or set/mutex (only Values exist).
func simpleSort(f *FieldOperation, keys []uint64) {
if keys != nil {
// keys might actually just point to record IDs, in which case, we don't
// want to shuffle the corresponding RecordIDs too, because that would just
// reverse our swaps. If they're different, we actually need to swap them
// both.
if &keys[0] != &f.RecordIDs[0] {
// sorting by record IDs
if f.Values != nil && f.Signed != nil {
if f.Values != nil && f.Signed != nil { // time quantum field
for i := 1; i < len(keys); i++ {
for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
keys[j-1], keys[j] = keys[j], keys[j-1]
@ -362,7 +421,7 @@ func simpleSort(f *FieldOperation, keys []uint64) {
}
}
} else if f.Values != nil {
} else if f.Values != nil { // set/mutex/bool
for i := 1; i < len(keys); i++ {
for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
keys[j-1], keys[j] = keys[j], keys[j-1]
@ -371,7 +430,7 @@ func simpleSort(f *FieldOperation, keys []uint64) {
f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1]
}
}
} else if f.Signed != nil {
} else if f.Signed != nil { // can't-happen, we think
for i := 1; i < len(keys); i++ {
for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
keys[j-1], keys[j] = keys[j], keys[j-1]
@ -379,7 +438,7 @@ func simpleSort(f *FieldOperation, keys []uint64) {
f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]
}
}
} else {
} else { // can't-happen, we think
for i := 1; i < len(keys); i++ {
for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
keys[j-1], keys[j] = keys[j], keys[j-1]
@ -388,7 +447,14 @@ func simpleSort(f *FieldOperation, keys []uint64) {
}
}
} else {
if f.Values != nil && f.Signed != nil {
if f.Values == nil && f.Signed != nil {
for i := 1; i < len(f.RecordIDs); i++ {
for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]
}
}
} else if f.Values != nil && f.Signed != nil { // can't happen, we think
for i := 1; i < len(f.RecordIDs); i++ {
for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
@ -396,22 +462,14 @@ func simpleSort(f *FieldOperation, keys []uint64) {
f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]
}
}
} else if f.Values != nil {
} else if f.Values != nil { // only happens during testing
for i := 1; i < len(f.RecordIDs); i++ {
for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1]
}
}
} else if f.Signed != nil {
for i := 1; i < len(f.RecordIDs); i++ {
for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]
}
}
} else {
// why do we only have record IDs? I don't know
} else { // should definitely not happen
for i := 1; i < len(f.RecordIDs); i++ {
for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
@ -493,12 +551,7 @@ func sortPartialByKeys(f *FieldOperation, keys []uint64, shift int) {
if end-start > 32 {
sortPartialByKeys(&bucketOp, keys[start:end], nextShift)
} else {
// naive stdlib sort
if externalKeys {
simpleSort(&bucketOp, keys[start:end])
} else {
simpleSort(&bucketOp, nil)
}
simpleSort(&bucketOp, keys[start:end])
}
}
}
@ -534,16 +587,15 @@ func (got *FieldOperation) Compare(expected *FieldOperation) error {
if expected == nil {
return nil
}
if len(expected.RecordIDs) == 0 && len(expected.Values) == 0 && len(expected.Signed) == 0 {
// We don't worry about non-empty Values or Signed here, because in theory
// RecordIDs are the Source of Truth as to what's in the op.
if len(expected.RecordIDs) == 0 {
return nil
}
return fmt.Errorf("expected field operation with %d records, got nil", len(expected.RecordIDs))
}
if expected == nil {
if got == nil {
return nil
}
if len(got.RecordIDs) == 0 && len(got.Values) == 0 && len(got.Signed) == 0 {
if len(got.RecordIDs) == 0 {
return nil
}
return fmt.Errorf("expected empty field operation, got %d records", len(got.RecordIDs))
@ -578,52 +630,6 @@ func (got *FieldOperation) Compare(expected *FieldOperation) error {
return nil
}
func translateUnsignedSlice(target []uint64, mapping []uint64) (err error) {
oops := 0
for i, v := range target {
if v >= uint64(len(mapping)) {
oops++
} else {
target[i] = mapping[v]
}
}
if oops > 0 {
return fmt.Errorf("encountered %d out-of-range keys when applying translation mapping", oops)
}
return nil
}
// TranslateUnsigned translates keys according to the provided mapping. This
// is used for sets, mutexes, and time quantums.
func (op *FieldOperation) TranslateKeys(mapping []uint64) error {
return translateUnsignedSlice(op.RecordIDs, mapping)
}
// TranslateUnsigned translates values according to the provided mapping. This
// is used for sets, mutexes, and time quantums.
func (op *FieldOperation) TranslateUnsigned(mapping []uint64) error {
return translateUnsignedSlice(op.Values, mapping)
}
// TranslateSigned translates signed values according to the provided mapping.
// If we're using this, it's because we're in an integer-type field, which
// admits using keys for fields, so all key values are actually non-negative,
// but the field's type still requires values be expressed as signed ints.
func (op *FieldOperation) TranslateSigned(mapping []uint64) error {
oops := 0
for i, v := range op.Signed {
if v >= int64(len(mapping)) {
oops++
} else {
op.Signed[i] = int64(mapping[v])
}
}
if oops > 0 {
return fmt.Errorf("encountered %d out-of-range signed values when applying translation mapping", oops)
}
return nil
}
// ShardOperations is a set of Operations associated with a specific shard.
type ShardOperations struct {
Shard uint64
@ -633,19 +639,17 @@ type ShardOperations struct {
// Request is a complete ingest request, which may be any combination
// of operations, which may apply to multiple shards.
type Request struct {
FieldTypes map[string]FieldType
Ops []*Operation
Ops []*Operation
}
// ShardedRequest is an ingest request, split up into individual per-shard
// operations.
type ShardedRequest struct {
FieldTypes map[string]FieldType
Ops map[uint64][]*Operation
Ops map[uint64][]*Operation
}
// ByShard converts a request into the same request, only sharded.
func (r *Request) ByShard() (*ShardedRequest, error) {
func (r *Request) ByShard(fields map[string]FieldType) (*ShardedRequest, error) {
if len(r.Ops) == 0 {
return &ShardedRequest{Ops: nil}, nil
}
@ -662,12 +666,12 @@ func (r *Request) ByShard() (*ShardedRequest, error) {
if op.OpType == OpClear || op.OpType == OpWrite || op.OpType == OpDelete {
sharded := ShardIDs(op.ClearRecordIDs)
for shard, data := range sharded {
shards[shard] = &Operation{OpType: op.OpType, ClearRecordIDs: data, ClearFields: op.ClearFields, FieldOps: map[string]*FieldOperation{}}
shards[shard] = &Operation{OpType: op.OpType, Seq: op.Seq, ClearRecordIDs: data, ClearFields: op.ClearFields, FieldOps: map[string]*FieldOperation{}}
}
}
for field, fieldOp := range op.FieldOps {
sharded := fieldOp.ByShard()
sorter := fieldTypeSorts[r.FieldTypes[field]]
sorter := fieldTypeSorts[fields[field]]
if sorter == nil {
sorter = (*FieldOperation).SortByRecords
}
@ -678,7 +682,7 @@ func (r *Request) ByShard() (*ShardedRequest, error) {
if op.OpType == OpWrite {
return nil, fmt.Errorf("write operation has field operation data (%d items) for shard %d, but no clear data", len(data.RecordIDs), shard)
}
shardOp = &Operation{OpType: op.OpType}
shardOp = &Operation{OpType: op.OpType, Seq: op.Seq}
shards[shard] = shardOp
shardOp.FieldOps = map[string]*FieldOperation{field: data}
} else {
@ -697,18 +701,139 @@ func (r *Request) ByShard() (*ShardedRequest, error) {
return &ShardedRequest{Ops: req}, nil
}
// merge combines the components of a sharded request back into a single
// unsharded request, processing shards in numerical order.
func (s *ShardedRequest) merge() *Request {
req := &Request{}
if s == nil || len(s.Ops) == 0 {
return req
}
shards := make([]uint64, 0, len(s.Ops))
for shard := range s.Ops {
shards = append(shards, shard)
}
sort.Slice(shards, func(i, j int) bool { return shards[i] < shards[j] })
for _, shard := range shards {
ops := s.Ops[shard]
for _, op := range ops {
var _ *Operation
if op.Seq >= len(req.Ops) {
// Pad out with nil *Operations to the required length
req.Ops = append(req.Ops, make([]*Operation, op.Seq+1-len(req.Ops))...)
}
if req.Ops[op.Seq] == nil {
req.Ops[op.Seq] = op.clone()
continue
}
req.Ops[op.Seq].merge(op)
}
}
return req
}
func (r *Request) Dump(logf func(string, ...interface{})) {
logf("req: %#v", r)
for _, op := range r.Ops {
logf("op: %#v", op)
if len(op.ClearRecordIDs) > 0 {
logf(" clearRecordIDs: %d", op.ClearRecordIDs)
if len(op.ClearRecordIDs) > 8 {
logf(" clearRecordIDs: %d...+%d", op.ClearRecordIDs[:8], len(op.ClearRecordIDs)-8)
} else {
logf(" clearRecordIDs: %d", op.ClearRecordIDs)
}
}
if len(op.ClearFields) > 0 {
logf(" clearFields: %s", op.ClearFields)
if len(op.ClearFields) > 8 {
logf(" clearFields: %s...+%d", op.ClearFields[:8], len(op.ClearFields)-8)
} else {
logf(" clearFields: %s", op.ClearFields)
}
}
for field, fieldOp := range op.FieldOps {
logf(" field %q: %#v", field, fieldOp)
if fieldOp != nil {
logf(" field %q: op (%d/%d/%d)", field, len(fieldOp.RecordIDs), len(fieldOp.Values), len(fieldOp.Signed))
if len(fieldOp.RecordIDs) > 0 {
if len(fieldOp.RecordIDs) > 8 {
logf(" records %d...+%d", fieldOp.RecordIDs[:8], len(fieldOp.RecordIDs)-8)
} else {
logf(" records %d", fieldOp.RecordIDs)
}
}
} else {
logf(" field %q: nil op", field)
}
}
}
}
func (r *Request) Compare(other *Request) error {
if other == nil {
if r != nil && len(r.Ops) != 0 {
return errors.New("non-empty sharded request can't equal empty/nil sharded request")
}
// empty and nil are allowed
return nil
}
if r == nil {
if other != nil && len(other.Ops) != 0 {
return errors.New("non-empty sharded request can't equal empty/nil sharded request")
}
// empty and nil are allowed
return nil
}
ops := r.Ops
ops2 := other.Ops
if len(ops2) != len(ops) {
return fmt.Errorf("expected %d ops, got %d", len(ops), len(ops2))
}
for i, op := range ops {
if err := op.Compare(ops2[i]); err != nil {
return fmt.Errorf("op %d: %v", i, err)
}
}
return nil
}
// Compare checks whether two ShardedRequest objects represent the same
// data. Empty shards shouldn't have entries in the map in the first place,
// so we don't accept a nil or 0-length slice of ops as equal to the
// shard key not existing, but we do accept nil or empty requests as
// equal to each other.
func (s *ShardedRequest) Compare(other *ShardedRequest) error {
if other == nil {
if s != nil && len(s.Ops) != 0 {
return errors.New("non-empty sharded request can't equal empty/nil sharded request")
}
// empty and nil are allowed
return nil
}
if s == nil {
if other != nil && len(other.Ops) != 0 {
return errors.New("non-empty sharded request can't equal empty/nil sharded request")
}
// empty and nil are allowed
return nil
}
for shard, ops := range s.Ops {
ops2, ok := other.Ops[shard]
if !ok {
return fmt.Errorf("shard %d missing in other", shard)
}
if len(ops2) != len(ops) {
return fmt.Errorf("shard %d: expected %d ops, got %d", shard, len(ops), len(ops2))
}
for i, op := range ops {
if err := op.Compare(ops2[i]); err != nil {
return fmt.Errorf("shard %d, op %d: %v", shard, i, err)
}
}
}
if len(other.Ops) != len(s.Ops) {
for shard := range other.Ops {
if _, ok := s.Ops[shard]; !ok {
return fmt.Errorf("shard %d missing in self", shard)
}
}
}
return nil
}

View file

@ -12,31 +12,29 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package ingest_test
package ingest
import (
"math/rand"
"reflect"
"testing"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/shardwidth"
)
type opShardingTestCase struct {
name string
input ingest.Request
output *ingest.ShardedRequest
input *Request
output *ShardedRequest
}
var opShardingTestCases = []opShardingTestCase{
{
name: "sample",
input: ingest.Request{
Ops: []*ingest.Operation{
input: &Request{
Ops: []*Operation{
{
OpType: ingest.OpSet,
FieldOps: map[string]*ingest.FieldOperation{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"shard0": {
RecordIDs: []uint64{0, 1},
},
@ -55,8 +53,9 @@ var opShardingTestCases = []opShardingTestCase{
},
},
{
OpType: ingest.OpRemove,
FieldOps: map[string]*ingest.FieldOperation{
OpType: OpRemove,
Seq: 1,
FieldOps: map[string]*FieldOperation{
"shard0-2": {
RecordIDs: []uint64{1, 2<<shardwidth.Exponent + 1},
},
@ -64,12 +63,12 @@ var opShardingTestCases = []opShardingTestCase{
},
},
},
output: &ingest.ShardedRequest{
Ops: map[uint64][]*ingest.Operation{
output: &ShardedRequest{
Ops: map[uint64][]*Operation{
0: {
{
OpType: ingest.OpSet,
FieldOps: map[string]*ingest.FieldOperation{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"shard0": {
RecordIDs: []uint64{0, 1},
},
@ -81,8 +80,9 @@ var opShardingTestCases = []opShardingTestCase{
},
},
{
OpType: ingest.OpRemove,
FieldOps: map[string]*ingest.FieldOperation{
OpType: OpRemove,
Seq: 1,
FieldOps: map[string]*FieldOperation{
"shard0-2": {
RecordIDs: []uint64{1},
},
@ -91,8 +91,8 @@ var opShardingTestCases = []opShardingTestCase{
},
1: {
{
OpType: ingest.OpSet,
FieldOps: map[string]*ingest.FieldOperation{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{
"shard0-1": {
RecordIDs: []uint64{
1 << shardwidth.Exponent,
@ -109,8 +109,9 @@ var opShardingTestCases = []opShardingTestCase{
},
2: {
{
OpType: ingest.OpRemove,
FieldOps: map[string]*ingest.FieldOperation{
OpType: OpRemove,
Seq: 1,
FieldOps: map[string]*FieldOperation{
"shard0-2": {
RecordIDs: []uint64{2<<shardwidth.Exponent + 1},
},
@ -122,15 +123,115 @@ var opShardingTestCases = []opShardingTestCase{
},
}
func TestOpSharding(t *testing.T) {
func TestOpShardingSmall(t *testing.T) {
codec, _ := NewJSONCodec(nil)
_ = codec.AddIntField("shard0", nil)
_ = codec.AddIntField("shard1", nil)
_ = codec.AddIntField("shard0-1", nil)
_ = codec.AddIntField("shard0-2", nil)
fieldTypes := codec.FieldTypes()
for _, c := range opShardingTestCases {
sharded, err := c.input.ByShard()
sharded, err := c.input.ByShard(fieldTypes)
if err != nil {
t.Errorf("sharding: unexpected error %v", err)
}
if !reflect.DeepEqual(sharded, c.output) {
t.Fatalf("%s: expected %#v, got %#v", c.name, c.output, sharded)
if err := sharded.Compare(c.output); err != nil {
t.Fatalf("%s: shard: %v", c.name, err)
}
merged := sharded.merge()
if err := c.input.Compare(merged); err != nil {
t.Fatalf("%s: merge: %v", c.name, err)
}
}
}
func TestOpShardingLarge(t *testing.T) {
codec, err := NewJSONCodec(nil)
if err != nil {
t.Fatalf("creating codec: %v", err)
}
_ = codec.AddSetField("set", nil)
_ = codec.AddTimeQuantumField("tq", nil)
_ = codec.AddIntField("int", nil)
// and now we populate encodeTests[1] with a larger pool of data
// if this isn't large enough, the scattering across shards means
// we coincidentally end up with the time quantum field not getting
// tested on simpleSort.
const dataSize = 120000
shardCount := uint64(300) // shards we want to target
passes := uint64(0)
recordIDs := make([]uint64, dataSize)
values := make([]uint64, dataSize)
timeStamps := make([]int64, dataSize)
signedValues := make([]int64, dataSize)
for i := uint64(0); i < dataSize; i++ {
if (i % shardCount) == 0 {
passes++
}
recordIDs[i] = ((i % shardCount) << shardwidth.Exponent) + passes
values[i] = (i % 4)
timeStamps[i] = int64(1234567890e9 + (i * 100e9))
signedValues[i] = (int64(i) % 16) // no negative values because they won't work with keys
}
op := &Operation{
OpType: OpSet,
FieldOps: map[string]*FieldOperation{},
}
req := &Request{
Ops: []*Operation{op},
}
op.FieldOps["tq"] = &FieldOperation{
RecordIDs: append([]uint64{}, recordIDs...),
Values: append([]uint64{}, values...),
Signed: timeStamps,
}
op.FieldOps["int"] = &FieldOperation{
RecordIDs: recordIDs,
Signed: signedValues,
}
// for sets, we want to shuffle things into fewer shards, and ensure
// non-duplication of values within each record, but also have lots
// of duplication of record IDs in the low shards
recordIDs = make([]uint64, dataSize)
values = make([]uint64, dataSize)
valuesPerRecord := uint64(5)
recordsPerShard := dataSize / valuesPerRecord / 30
if recordsPerShard < 1 {
recordsPerShard = 1
}
shard := uint64(0)
nextID := uint64(0)
nextValue := uint64(0)
for i := uint64(0); i < dataSize; i++ {
recordIDs[i] = nextID
values[i] = nextValue + (i % valuesPerRecord)
nextValue++
if nextValue == valuesPerRecord {
nextValue = 0
nextID++
if nextID%(1<<shardwidth.Exponent) == recordsPerShard {
shard++
nextID = (shard << shardwidth.Exponent)
if valuesPerRecord > 1 {
valuesPerRecord--
}
}
}
}
op.FieldOps["set"] = &FieldOperation{
RecordIDs: recordIDs,
Values: values,
}
fieldTypes := codec.FieldTypes()
sharded, err := req.ByShard(fieldTypes)
if err != nil {
t.Errorf("sharding: unexpected error %v", err)
}
merged := sharded.merge()
if err := req.Compare(merged); err != nil {
t.Fatalf("merge comparison: %v", err)
}
}
@ -139,7 +240,7 @@ func TestFancySharding(t *testing.T) {
const recordCount = 5000
grr := rand.New(rand.NewSource(0))
for i := 0; i < 100; i++ {
f := &ingest.FieldOperation{RecordIDs: make([]uint64, recordCount), Values: make([]uint64, recordCount)}
f := &FieldOperation{RecordIDs: make([]uint64, recordCount), Values: make([]uint64, recordCount)}
shards := make([]int, shardLimit)
for j := range f.RecordIDs {
v := uint64(grr.Int63n(shardLimit << shardwidth.Exponent))

View file

@ -1,15 +0,0 @@
// Copyright 2021 Molecula Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package ingest

137
ingest/translate_test.go Normal file
View file

@ -0,0 +1,137 @@
// Copyright 2021 Molecula Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package ingest
import (
"fmt"
"testing"
)
// stableTranslator implements a key translator that can be reused and
// will continue to give the same keys for the same values. Possibly
// surprisingly, it will invent new keys for IDs it is asked about but
// hasn't seen. This allows us to give a codec which would use keys on
// translation a request which contains arbitrary numbers, and request
// text that would parse into that request.
type stableTranslator struct {
in map[string]uint64
out map[uint64]string
next uint64
}
func (s *stableTranslator) TranslateKeys(keys ...string) (map[string]uint64, error) {
ret := make(map[string]uint64, len(keys))
for _, key := range keys {
if existing, ok := s.in[key]; ok {
ret[key] = existing
continue
}
id := s.next
// but what if someone already translated that ID, so now it already
// exists?
if _, ok := s.out[id]; ok {
for k := range s.out {
if k > id {
id = k
}
}
// one larger than the largest we already have. this could
// wrap around, in which case, it's your own fault.
id++
}
s.next = id + 1
s.in[key] = id
s.out[id] = key
ret[key] = id
}
return ret, nil
}
func (s *stableTranslator) TranslateIDs(ids ...uint64) (map[uint64]string, error) {
ret := make(map[uint64]string, len(ids))
for _, id := range ids {
if existing, ok := s.out[id]; ok {
ret[id] = existing
continue
}
key := fmt.Sprintf("k-%d", id)
s.in[key] = id
s.out[id] = key
ret[id] = key
if id >= s.next {
s.next = id + 1
}
}
return ret, nil
}
// newStableTranslator produces a translator which can translate forwards
// and backwards and invent new things if it needs to. Don't use this.
func newStableTranslator() *stableTranslator {
return &stableTranslator{
in: make(map[string]uint64),
out: make(map[uint64]string),
}
}
func TestTranslateReuse(t *testing.T) {
// the original stable-translator design had a flaw in that it
// assumed that each new ID would always come from a string translation,
// never from a key translation, and that they'd show up sequentially.
tr := newStableTranslator()
orig, err := tr.TranslateIDs(1, 2)
tr.next = 1 // intentionally break the translation logic for testing purposes
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
var s [6]string
stash := s[:0]
for _, v := range orig {
stash = append(stash, v)
}
for i := range s[len(orig):] {
stash = append(stash, fmt.Sprintf("key-%d", i))
}
keys, err := tr.TranslateKeys(stash...)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
ids := make([]uint64, 0, len(keys))
for _, v := range keys {
ids = append(ids, v)
}
idMap, err := tr.TranslateIDs(ids...)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
for k, v := range keys {
if idMap[v] != k {
t.Fatalf("translate mismatch: keys %q->%d, ids %d->%q",
k, v, v, idMap[v])
}
}
for id, key := range idMap {
if keys[key] != id {
t.Fatalf("translate mismatch: ids %d->%q, keys %q->%d",
id, key, key, keys[key])
}
}
for id, key := range orig {
if keys[key] != id {
t.Fatalf("translate mismatch: original ids %d->%q, keys %q->%d",
id, key, key, keys[key])
}
}
}

View file

@ -19,8 +19,6 @@ import (
"reflect"
"strconv"
"unsafe"
"github.com/pkg/errors"
)
// StringTable is a mapping of strings to temporary IDs.
@ -85,11 +83,8 @@ func (tbl *StringTable) IntID(in []byte) (int64, error) {
// integers and a translation function from strings to "real" keys, yields
// a translation/lookup slice. If it cannot translate all the keys, it
// returns an error.
//
// The lookup function corresponds to the FindKeys/CreateKeys methods of
// featurebase translators, by an AMAZING coincidence.
func MapForStringTable(tbl *StringTable, lookup func(...string) (map[string]uint64, error)) ([]uint64, error) {
lookedUp, err := lookup(tbl.names...)
func (tbl *StringTable) MakeIDMap(keys KeyTranslator) ([]uint64, error) {
lookedUp, err := keys.TranslateKeys(tbl.names...)
if err != nil {
return nil, err
}
@ -103,22 +98,38 @@ func MapForStringTable(tbl *StringTable, lookup func(...string) (map[string]uint
return out, nil
}
// TimeFormatForUnit returns the time transfer format (between the update encoder and the update applier) with appropriate resolution for a quantum unit.
func TimeFormatForUnit(unit rune) string {
switch unit {
case 'Y':
return "2006"
case 'M':
return "200601"
case 'D':
return "20060102"
case 'H':
return "2006010203"
default:
panic(errors.Errorf("invalid quantum unit: %q", unit))
// translateSigned replaces values from 0 to len(mapping)-1 with the
// elements of mapping. It yields an error if any values aren't
// mapped.
func translateSigned(mapping []uint64, values []int64) error {
oops := 0
for i, v := range values {
if v >= int64(len(mapping)) {
oops++
} else {
values[i] = int64(mapping[v])
}
}
if oops > 0 {
return fmt.Errorf("encountered %d out-of-range signed values when applying translation mapping", oops)
}
return nil
}
// TODO: bool
// TODO: timestamp (just sugar on top of IntVector)
// translateUnsigned replaces values from 0 to len(mapping)-1 with the
// elements of mapping. It yields an error if any values aren't
// mapped.
func translateUnsigned(mapping []uint64, values []uint64) error {
oops := 0
for i, v := range values {
if v >= uint64(len(mapping)) {
oops++
} else {
values[i] = mapping[v]
}
}
if oops > 0 {
return fmt.Errorf("encountered %d out-of-range signed values when applying translation mapping", oops)
}
return nil
}

114
ingest/vec_test.go Normal file
View file

@ -0,0 +1,114 @@
// Copyright 2021 Molecula Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package ingest
import (
"errors"
"testing"
)
type badTranslator struct{}
func (b badTranslator) TranslateKeys(keys ...string) (map[string]uint64, error) {
if len(keys) == 0 {
return nil, errors.New("no keys")
}
m := make(map[string]uint64)
skip := true
for i, k := range keys {
if skip {
skip = false
continue
}
m[k] = uint64(i)
}
out := make([]uint64, len(keys)-1)
for i := range out {
out[i] = uint64(i)
}
return nil, nil
}
func (b badTranslator) TranslateIDs(...uint64) (map[uint64]string, error) {
return nil, nil
}
func TestStringTableErrors(t *testing.T) {
tbl := NewStringTable()
btr := badTranslator{}
_, keyErr := tbl.MakeIDMap(btr)
if keyErr == nil {
t.Fatalf("expected error passed up from failed translate, didn't get it")
}
a1, err := tbl.ID([]byte("a"))
if err != nil {
t.Fatalf("getting translation for key: %v", err)
}
b1, err := tbl.ID([]byte("b"))
if err != nil {
t.Fatalf("getting translation for key: %v", err)
}
_, keyErr = tbl.MakeIDMap(btr)
if keyErr == nil {
t.Fatalf("expected error for short translate, didn't get it")
}
tr := newStableTranslator()
_, err = tr.TranslateKeys("c", "d")
if err != nil {
t.Fatalf("translating stray keys: %v", err)
}
m, err := tbl.MakeIDMap(tr)
if err != nil {
t.Fatalf("creating lookup: %v", err)
}
var y = []uint64{a1, b1}
err = translateUnsigned(m, y)
if err != nil {
t.Fatalf("unexpected unsigned translation error: %v", err)
}
trResults, err := tr.TranslateKeys("a", "b")
if err != nil {
t.Fatalf("unexpected translation error: %v", err)
}
if y[0] != trResults["a"] {
t.Fatalf("expected %d, got %d", trResults["a"], y[0])
}
if y[1] != trResults["b"] {
t.Fatalf("expected %d, got %d", trResults["b"], y[1])
}
y[0] = a1
y[1] = (a1 + b1 + 1) // assumed not to be any of them
err = translateUnsigned(m, y)
if err == nil {
t.Fatalf("no error from translating invalid table")
}
z := []int64{int64(a1), int64(b1)}
err = translateSigned(m, z)
if err != nil {
t.Fatalf("unexpected unsigned translation error: %v", err)
}
if uint64(z[0]) != trResults["a"] {
t.Fatalf("expected %d, got %d", trResults["a"], z[0])
}
if uint64(z[1]) != trResults["b"] {
t.Fatalf("expected %d, got %d", trResults["b"], z[1])
}
z[0] = int64(a1)
z[1] = int64(a1 + b1 + 1) // assumed not to be any of them
err = translateSigned(m, z)
if err == nil {
t.Fatalf("no error from translating invalid table")
}
}

View file

@ -26,9 +26,7 @@ import (
"testing"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/server"
"github.com/molecula/featurebase/v2/test"
"github.com/pkg/errors"
)
@ -448,14 +446,7 @@ func TestIngestTestcases(t *testing.T) {
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
c := test.MustRunCluster(t, 1,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerNodeID("node0"),
pilosa.OptServerClusterHasher(&offsetModHasher{}),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
)},
)
c := test.MustRunCluster(t, 3)
defer c.Close()
coord := c.GetPrimary()

File diff suppressed because it is too large Load diff

View file

@ -234,4 +234,25 @@ message ResizeAbortMessage {
message ResizeNodeMessage {
string NodeID = 1;
string Action = 2;
}
}
message FieldOperation {
repeated uint64 RecordIDs = 1;
repeated uint64 Values = 2;
repeated int64 Signed = 3;
}
message ShardIngestOperation {
string OpType = 1;
repeated uint64 ClearRecordIDs = 2;
repeated string ClearFields = 3;
map <string, FieldOperation> FieldOps = 4;
}
message ShardIngestOperations {
repeated ShardIngestOperation Ops = 1;
}
message ShardedIngestRequest {
map <uint64, ShardIngestOperations> Ops = 1;
}

View file

@ -5980,10 +5980,7 @@ func (m *Row) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -6068,10 +6065,7 @@ func (m *RowMatrix) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -6194,10 +6188,7 @@ func (m *SignedRow) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -6356,10 +6347,7 @@ func (m *RowIdentifiers) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -6486,10 +6474,7 @@ func (m *IDList) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -6593,10 +6578,7 @@ func (m *ExtractedIDColumn) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -6713,10 +6695,7 @@ func (m *ExtractedIDMatrix) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -6799,10 +6778,7 @@ func (m *KeyList) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7016,10 +6992,7 @@ func (m *ExtractedTableValue) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7156,10 +7129,7 @@ func (m *ExtractedTableColumn) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7274,10 +7244,7 @@ func (m *ExtractedTableField) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7396,10 +7363,7 @@ func (m *ExtractedTable) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7520,10 +7484,7 @@ func (m *Pair) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7642,10 +7603,7 @@ func (m *PairField) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7762,10 +7720,7 @@ func (m *PairsField) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -7835,10 +7790,7 @@ func (m *Int64) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -8008,10 +7960,7 @@ func (m *FieldRow) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -8134,10 +8083,7 @@ func (m *GroupCount) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -8273,10 +8219,7 @@ func (m *ValCount) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -8365,10 +8308,7 @@ func (m *Decimal) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -8620,10 +8560,7 @@ func (m *QueryRequest) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -8740,10 +8677,7 @@ func (m *QueryResponse) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -9356,10 +9290,7 @@ func (m *QueryResult) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -9843,10 +9774,7 @@ func (m *ImportRequest) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -10308,10 +10236,7 @@ func (m *ImportValueRequest) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -10481,10 +10406,7 @@ func (m *AtomicRecord) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -10567,10 +10489,7 @@ func (m *AtomicImportResponse) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -10737,10 +10656,7 @@ func (m *TranslateKeysRequest) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -10867,10 +10783,7 @@ func (m *TranslateKeysResponse) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -11061,10 +10974,7 @@ func (m *TranslateIDsRequest) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -11147,10 +11057,7 @@ func (m *TranslateIDsResponse) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -11267,10 +11174,7 @@ func (m *ImportRoaringRequestView) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -11484,10 +11388,7 @@ func (m *ImportRoaringRequest) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
@ -11604,10 +11505,7 @@ func (m *GroupCounts) Unmarshal(dAtA []byte) error {
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) < 0 {
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {

View file

@ -23,6 +23,7 @@ import (
"sort"
"sync"
"github.com/molecula/featurebase/v2/ingest"
"github.com/molecula/featurebase/v2/topology"
"github.com/pkg/errors"
)
@ -98,6 +99,39 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
ReadFrom(io.Reader) (int64, error)
}
// This implements ingest's key translator interface, which differs
// slightly because we want to be able to do fast lookups on arbitrary
// IDs which are not necessarily contiguous small values, so the []string
// from TranslateIDs isn't a good fit.
type ingestKeyTranslator struct {
store TranslateStore
}
var _ ingest.KeyTranslator = &ingestKeyTranslator{}
func (i ingestKeyTranslator) TranslateKeys(keys ...string) (map[string]uint64, error) {
return i.store.CreateKeys(keys...)
}
func (i ingestKeyTranslator) TranslateIDs(ids ...uint64) (map[uint64]string, error) {
keys, err := i.store.TranslateIDs(ids)
if err != nil {
return nil, err
}
if len(keys) != len(ids) {
return nil, fmt.Errorf("translating %d id(s), got %d key(s)", len(ids), len(keys))
}
out := make(map[uint64]string, len(keys))
for i, id := range ids {
out[id] = keys[i]
}
return out, nil
}
func newIngestKeyTranslatorFromStore(s TranslateStore) *ingestKeyTranslator {
return &ingestKeyTranslator{store: s}
}
// TranslatorSummary is returned, for example from the boltdb string key translators,
// by calling ComputeTranslatorSummary(). Non-boltdb mocks, etc no-op that method.
type TranslatorSummary struct {