diff --git a/api.go b/api.go index bbdb8c6b4..b5d2f83c4 100644 --- a/api.go +++ b/api.go @@ -35,6 +35,7 @@ import ( "time" "github.com/molecula/featurebase/v2/disco" + "github.com/molecula/featurebase/v2/ingest" "github.com/molecula/featurebase/v2/pql" "github.com/molecula/featurebase/v2/roaring" "github.com/molecula/featurebase/v2/stats" @@ -1815,6 +1816,180 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu return nil } +func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string, stream io.Reader) error { + span, _ := tracing.StartSpanFromContext(ctx, "API.IngestOperations") + defer span.Finish() + + if err := api.validate(apiIngestOperations); err != nil { + return errors.Wrap(err, "validating api method") + } + + // Find the Index. + 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() + var lookup ingest.KeyLookupFunc + if index.Keys() { + lookup = func(keys ...string) (map[string]uint64, error) { + return api.cluster.createIndexKeys(ctx, indexName, keys...) + } + } + codec, err := ingest.NewJSONCodec(lookup) + if err != nil { + return errors.Wrap(err, "creating JSON codec") + } + knownFields := map[string]*Field{} + for _, field := range fields { + var lookup ingest.KeyLookupFunc + if field.usesKeys { + lookup = field.translateStore.CreateKeys + } + knownFields[field.name] = field + switch field.Type() { + case "set": + if err = codec.AddSetField(field.name, lookup); err != nil { + return fmt.Errorf("adding set field to codec: %w", err) + } + case "time": + if err = codec.AddTimeQuantumField(field.name, lookup); err != nil { + return fmt.Errorf("adding time field to codec: %w", err) + } + case "mutex": + if err = codec.AddMutexField(field.name, lookup); err != nil { + return fmt.Errorf("adding mutex field to codec: %w", err) + } + case "bool": + if err = codec.AddBoolField(field.name); err != nil { + return fmt.Errorf("adding mutex field to codec: %w", err) + } + case "int": + if err = codec.AddIntField(field.name, lookup); err != nil { + return fmt.Errorf("adding mutex field to codec: %w", err) + } + case "decimal": + if err = codec.AddDecimalField(field.name, field.options.Scale); err != nil { + return fmt.Errorf("adding mutex field to codec: %w", err) + } + case "timestamp": + nanos := TimeUnitNanos(field.options.TimeUnit) + if err = codec.AddTimestampField(field.name, time.Duration(nanos), field.options.Base); err != nil { + return fmt.Errorf("adding mutex field to codec: %w", err) + } + default: + return fmt.Errorf("unhandled field type %q", field.Type()) + } + } + req, err := codec.Parse(stream) + if err != nil { + return errors.Wrap(err, "parsing input data") + } + sharded, err := req.Shard() + if err != nil { + return errors.Wrap(err, "sharding input data") + } + eg, ctx := errgroup.WithContext(ctx) + 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) + }) + } + return eg.Wait() +} + +// applyOperations applies a set of operations to one specific shard. +func (api *API) applyOperations(ctx context.Context, qcx *Qcx, index *Index, shard uint64, fields map[string]*Field, ops []*ingest.Operation) error { + // For each operation, we may have a set of records/fields to clear, and then + // also a set of fields to set/remove specific bits in. + opts := &ImportOptions{Presorted: true, IgnoreKeyCheck: true} + funcOpts := []ImportOption{OptImportOptionsIgnoreKeyCheck(true), OptImportOptionsPresorted(true), OptImportOptionsClear(true)} + for _, op := range ops { + // ClearRecordIDs should exist only for delete, clear, and write. For clear and write, + // we'll have a list of fields, for delete, it should be all the fields. + if len(op.ClearRecordIDs) > 0 { + // anonymous func lets us defer a finisher from any of the inner error returns + err := func() (e0 error) { + // WARNING: Depends on GetTx being per-shard/index, not per-field. + tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: index, Shard: shard}) + if err != nil { + return fmt.Errorf("getting Tx: %w", err) + } + defer finisher(&e0) + // For a delete, we don't look at the fields the codec was defined with, + // We delete from the existence field unconditionally and other fields + // if we know they exist. + if op.OpType == ingest.OpDelete { + err = clearExistenceColumns(qcx, index, op.ClearRecordIDs, shard) + if err != nil { + return fmt.Errorf("clearing existence columns: %w", err) + } + for name, field := range fields { + if err = field.ClearBits(tx, shard, op.ClearRecordIDs...); err != nil { + return fmt.Errorf("clearing field %q: %w", name, err) + } + } + return nil + } + // clear things that we need to wipe out, whether it's because + // this is a Clear op, or because it's a write op that + // specifies clears for the fields it's going to write to. + if len(op.ClearFields) > 0 { + for _, fieldName := range op.ClearFields { + field, ok := fields[fieldName] + if !ok { + return fmt.Errorf("can't find a field named %q", fieldName) + } + if err = field.ClearBits(tx, shard, op.ClearRecordIDs...); err != nil { + return fmt.Errorf("clearing record IDs: %w", err) + } + } + } + return nil + }() + if err != nil { + return err + } + } + opts.Clear = (op.OpType == ingest.OpRemove) + // reslice this rather than regenerating it. i'm so efficient. + if opts.Clear { + funcOpts = funcOpts[:3] + } else { + funcOpts = funcOpts[:2] + } + // for "set" and "write" ops, we'll be setting bits, for + // "remove" ops we'll be clearing them, and for "clear" ops + // there shouldn't be anything here. + for fieldName, fieldOp := range op.FieldOps { + field, ok := fields[fieldName] + if !ok { + return fmt.Errorf("can't find a field named %q", fieldName) + } + var err error + err = importExistenceColumns(qcx, index, fieldOp.RecordIDs, shard) + if err != nil { + return errors.Wrap(err, "importing existence columns") + } + switch field.Type() { + case "set", "time", "mutex": + err = field.Import(qcx, fieldOp.Values, fieldOp.RecordIDs, fieldOp.Signed, shard, funcOpts...) + case "int", "timestamp", "decimal": + err = field.importValue(qcx, fieldOp.RecordIDs, fieldOp.Signed, shard, opts) + default: + err = fmt.Errorf("unhandled field type %q", field.Type()) + } + if err != nil { + return err + } + } + } + return nil +} + func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error { ef := index.existenceField() if ef == nil { @@ -1831,6 +2006,22 @@ func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard ui return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard) } +func clearExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error { + ef := index.existenceField() + if ef == nil { + return nil + } + + existenceRowIDs := make([]uint64, len(columnIDs)) + // If we don't gratuitously hand-duplicate things in field.Import, + // the fact that fragment.bulkImport rewrites its row and column + // lists can burn us if we don't make a copy before doing the + // existence field write. + columnCopy := make([]uint64, len(columnIDs)) + copy(columnCopy, columnIDs) + return ef.Import(qcx, existenceRowIDs, columnCopy, nil, shard, OptImportOptionsClear(true)) +} + // ShardDistribution returns an object representing the distribution of shards // across nodes for each index, distinguishing between primary and replica. // The structure of this information is [indexName][nodeID][primaryOrReplica][]uint64. @@ -2513,6 +2704,7 @@ const ( apiIDCommit apiIDReset apiPartitionNodes + apiIngestOperations ) var methodsCommon = map[apiMethod]struct{}{ @@ -2580,4 +2772,5 @@ var methodsNormal = map[apiMethod]struct{}{ apiIDCommit: {}, apiIDReset: {}, apiPartitionNodes: {}, + apiIngestOperations: {}, } diff --git a/api_test.go b/api_test.go index 201a7c89a..17c1761d8 100644 --- a/api_test.go +++ b/api_test.go @@ -15,6 +15,7 @@ package pilosa_test import ( + "bytes" "context" "crypto/rand" "errors" @@ -411,6 +412,159 @@ func TestAPI_ImportValue(t *testing.T) { }) } +func TestAPI_Ingest(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)), + )}, + ) + defer c.Close() + + coord := c.GetPrimary() + // m0 := c.GetNode(0) + // m1 := c.GetNode(1) + // m2 := c.GetNode(2) + + index := "ingest" + setField := "set" + timeField := "tq" + + _, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = coord.API.CreateField(ctx, index, setField, pilosa.OptFieldTypeSet("none", 0)) + if err != nil { + t.Fatalf("creating field: %v", err) + } + _, err = coord.API.CreateField(ctx, index, timeField, pilosa.OptFieldTypeTime("YMD")) + if err != nil { + t.Fatalf("creating field: %v", err) + } + sampleJson := []byte(` +[ + { + "action": "set", + "records": { + "2": { + "set": [2], + "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [6] } + }, + "5": { "set": [3] }, + "8": { "set": [3] }, + "1": { + "set": [2], + "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] } + }, + "4": { "set": [3, 7] } + } + }, + { + "action": "clear", + "record_ids": [ 5, 6, 7 ], + "fields": [ "tq", "set" ] + }, + { + "action": "write", + "records": { + "8": { "tq": { "time": "2006-01-02T15:04:05.999999999Z", "values": [3, 4] } }, + "9": { "set": [7, 3] } + } + }, + { + "action": "delete", + "record_ids": [ 9 ] + } +] +`) + // just for set row 3: + // first operation should set it for 4, 5, and 8. + // clear operation should clear it for 5, 6, and 7, leaving it still set for 4 and 8. + // the write operation should clear set for record 8, even though record 8 doesn't + // contain that field in that op, because set is present in record 9, which also + // gets row 3 set. but then we delete 9. + // so after all that we expect Row(set=3) to be 4... + sampleBuf := bytes.NewBuffer(sampleJson) + qcx := coord.API.Txf().NewQcx() + defer func() { + if err := qcx.Finish(); err != nil { + t.Fatalf("finishing qcx: %v", err) + } + }() + err = coord.API.IngestOperations(ctx, qcx, index, sampleBuf) + if err != nil { + t.Fatalf("importing data: %v", err) + } + query := "Row(set=3)" + res, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query}) + if err != nil { + t.Errorf("query: %v", err) + } + r := res.Results[0].(*pilosa.Row).Columns() + if len(r) != 1 || r[0] != 4 { + t.Fatalf("expected row with 4 set, got %d", r) + } +} + +func BenchmarkIngest(b *testing.B) { + b.StopTimer() + buf := &bytes.Buffer{} + buf.WriteString(`[{"action": "write", "records": {`) + comma := "" + for i := 0; i < 1000000; i++ { + fmt.Fprintf(buf, `%s"%d": { "set": [%d, %d] }`, comma, i, i%2, (i%4)+2) + comma = ", " + } + buf.WriteString(`}}]`) + data := buf.Bytes() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c := test.MustRunCluster(b, 1, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + )}, + ) + defer c.Close() + + coord := c.GetPrimary() + m0 := c.GetNode(0) + // m1 := c.GetNode(1) + // m2 := c.GetNode(2) + + index := "ingest" + setField := "set" + _, err := coord.API.CreateIndex(ctx, index, pilosa.IndexOptions{Keys: false}) + if err != nil { + b.Fatalf("creating index: %v", err) + } + _, err = coord.API.CreateField(ctx, index, setField, pilosa.OptFieldTypeSet("none", 0)) + if err != nil { + b.Fatalf("creating field: %v", err) + } + b.ReportAllocs() + b.StartTimer() + for i := 0; i < b.N; i++ { + qcx := m0.API.Txf().NewQcx() + defer qcx.Abort() + err = coord.API.IngestOperations(ctx, qcx, index, bytes.NewBuffer(data)) + if err != nil { + b.Fatalf("ingest: %v", err) + } + err = qcx.Finish() + if err != nil { + b.Fatalf("finish: %v", err) + } + } +} + // offsetModHasher represents a simple, mod-based hashing offset by 1. type offsetModHasher struct{} diff --git a/field.go b/field.go index 5f147d735..a1c495d68 100644 --- a/field.go +++ b/field.go @@ -1178,6 +1178,29 @@ func (f *Field) ClearBit(tx Tx, rowID, colID uint64) (changed bool, err error) { return changed, nil } +// ClearBits clears all bits corresponding to the given record IDs in standard +// or BSI views. It does not delete bits from time quantum views. +func (f *Field) ClearBits(tx Tx, shard uint64, recordIDs ...uint64) error { + bsig := f.bsiGroup(f.name) + var v *view + if bsig != nil { + // looks like we're a BSI field? + v = f.view(viewBSIGroupPrefix + f.name) + } else { + v = f.view(viewStandard) + } + // it's fine if we never actually created the view, that means the + // bits are all clear! + if v == nil { + return nil + } + frag := v.Fragment(shard) + if frag == nil { + return nil + } + return frag.ClearRecords(tx, recordIDs) +} + func groupCompare(a, b string, offset int) (lt, eq bool) { if len(a) > offset { a = a[:offset] diff --git a/fragment.go b/fragment.go index 8b7f55f46..cf4c1c9f1 100644 --- a/fragment.go +++ b/fragment.go @@ -1741,7 +1741,7 @@ func (f *fragment) pos(rowID, columnID uint64) (uint64, error) { // Return an error if the column ID is out of the range of the fragment's shard. minColumnID := f.shard * ShardWidth if columnID < minColumnID || columnID >= minColumnID+ShardWidth { - return 0, errors.Errorf("column:%d out of bounds", columnID) + return 0, errors.Errorf("column:%d out of bounds for shard %d", columnID, f.shard) } return pos(rowID, columnID), nil } @@ -2581,6 +2581,32 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { return errors.Wrap(f.importPositions(tx, toSet, toClear, rowSet), "importing positions") } +// ClearRecords deletes all bits for the given records. It's basically +// the remove-only part of setting a mutex. +func (f *fragment) ClearRecords(tx Tx, recordIDs []uint64) error { + f.mu.Lock() + defer f.mu.Unlock() + + // create a mask of columns we care about + columns := roaring.NewSliceBitmap(recordIDs...) + + // we now need to find existing rows for these bits. + rowSet := make(map[uint64]struct{}) + var toClear []uint64 + callback := func(pos uint64) error { + toClear = append(toClear, pos) + rowID := pos / ShardWidth + rowSet[rowID] = struct{}{} + return nil + } + findExisting := roaring.NewBitmapBitmapFilter(columns, callback) + err := tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, findExisting) + if err != nil { + return errors.Wrap(err, "finding existing positions") + } + return errors.Wrap(f.importPositions(tx, nil, toClear, rowSet), "clearing records") +} + // importValue bulk imports a set of range-encoded values. func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDepth uint64, clear bool) error { f.mu.Lock() diff --git a/go.mod b/go.mod index 02133467c..e590bdd1b 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/HdrHistogram/hdrhistogram-go v1.1.0 // indirect github.com/beevik/ntp v0.3.0 github.com/benbjohnson/immutable v0.3.0 + github.com/buger/jsonparser v1.1.1 github.com/cespare/xxhash v1.1.0 github.com/davecgh/go-spew v1.1.1 github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect diff --git a/go.sum b/go.sum index f89f91301..5f4d1be61 100644 --- a/go.sum +++ b/go.sum @@ -42,6 +42,8 @@ github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0= github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs= github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84= +github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs= +github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko= github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc= diff --git a/http/client.go b/http/client.go index ce31012d7..509ade5ee 100644 --- a/http/client.go +++ b/http/client.go @@ -200,6 +200,61 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error } return rsp.Indexes, nil } + +// IngestSchema uses the new schema ingest endpoint. +func (c *InternalClient) IngestSchema(ctx context.Context, uri *pnet.URI, buf []byte) error { + if uri == nil { + uri = c.defaultURI + } + u := uri.Path("/internal/schema") + 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/json") + req.Header.Set("Accept", "application/json") + 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 +} + +// IngestOperations uses the new ingest endpoint for ingest data +func (c *InternalClient) IngestOperations(ctx context.Context, uri *pnet.URI, indexName string, buf []byte) error { + if uri == nil { + uri = c.defaultURI + } + u := uri.Path(fmt.Sprintf("/internal/ingest/%s", indexName)) + 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/json") + req.Header.Set("Accept", "application/json") + 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 +} + func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *pilosa.Schema, remote bool) error { u := uri.Path(fmt.Sprintf("/schema?remote=%v", remote)) buf, err := json.Marshal(s) diff --git a/ingest/codec.go b/ingest/codec.go new file mode 100644 index 000000000..a0ab071cc --- /dev/null +++ b/ingest/codec.go @@ -0,0 +1,561 @@ +// 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" + "io" + "io/ioutil" + "math" + "strconv" + "time" + + "github.com/buger/jsonparser" + "github.com/pkg/errors" +) + +// Featurebase has the following field types as of this writing, we plan to +// support all of them but not all are implemented. +// +// Type Single Signed Timestamp +// set no no no +// time no no yes +// mutex yes no no +// int yes yes no +// decimal yes yes no +// timestamp yes yes no + +type KeyLookupFunc func(...string) (map[string]uint64, 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 + AddBoolField(name string) error + AddIntField(name string, lookup KeyLookupFunc) error + AddDecimalField(name string, scale int64) error + AddTimestampField(name string, scale time.Duration, epoch int64) error + + // Parse data from a reader into the vectors. + // This must only be called once on a codec. + Parse(io.Reader) (*Request, error) +} + +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 + +type jsonFieldCodec struct { + valueKeys *StringTable + currentOp *FieldOperation + decode jsonDecFn + translate applyTranslationFn + // 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. + scaleUnit int64 + epoch int64 // used only by Timestamp fields + scratch []uint64 // reusable scratch space for sets of values + lookup KeyLookupFunc +} + +// 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 +} + +var _ Codec = &JSONCodec{} + +func NewJSONCodec(lookup KeyLookupFunc) (*JSONCodec, error) { + j := &JSONCodec{fields: map[string]*jsonFieldCodec{}} + if lookup != nil { + j.recKeys = NewStringTable() + j.keyLookup = lookup + } + return j, nil +} + +func (codec *JSONCodec) AddTimeQuantumField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeTimeQuantumValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateUnsigned + } + return codec.addField(name, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddSetField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeSetValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateUnsigned + } + return codec.addField(name, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddIntField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeIntValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateSigned + } + return codec.addField(name, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddMutexField(name string, lookup KeyLookupFunc) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeMutexValue + if lookup != nil { + fieldCodec.translate = (*FieldOperation).TranslateUnsigned + } + return codec.addField(name, fieldCodec, lookup) +} + +func (codec *JSONCodec) AddBoolField(name string) error { + fieldCodec := &jsonFieldCodec{} + fieldCodec.decode = fieldCodec.DecodeBoolValue + return codec.addField(name, fieldCodec, nil) +} + +// 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. +func (codec *JSONCodec) AddTimestampField(name string, timeScale time.Duration, epoch int64) error { + fieldCodec := &jsonFieldCodec{scaleUnit: int64(timeScale), epoch: epoch} + fieldCodec.decode = fieldCodec.DecodeTimeValue + return codec.addField(name, fieldCodec, nil) +} + +// 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.decode = fieldCodec.DecodeDecimalValue + return codec.addField(name, fieldCodec, nil) +} + +func (codec *JSONCodec) addField(name string, fieldCodec *jsonFieldCodec, lookup KeyLookupFunc) error { + if _, ok := codec.fields[name]; ok { + return fmt.Errorf("duplicate field %q", name) + } + if lookup != nil { + fieldCodec.valueKeys = NewStringTable() + fieldCodec.lookup = lookup + } + codec.fields[name] = fieldCodec + 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) { + 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 + } + }) + if err != nil { + return err + } + if arrayErr != nil { + return arrayErr + } + case jsonparser.String: + id, err := j.valueKeys.ID(data) + if err != nil { + return err + } + return cb(id) + case jsonparser.Number: + if j.valueKeys != nil { + return errors.New("expecting key, got numeric value") + } + id, err := strconv.ParseUint(pretendByteIsString(data), 10, 64) + if err != nil { + return err + } + return cb(id) + default: + return fmt.Errorf("expecting 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) { + return j.decodeSetOrValue(dataType, data, func(id uint64) error { + j.currentOp.AddPair(recID, id) + 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 { + switch dataType { + case jsonparser.String: + value, err := j.valueKeys.IntID(data) + if err != nil { + return err + } + j.currentOp.AddSignedPair(recID, value) + case jsonparser.Number: + if j.valueKeys != nil { + return errors.New("expecting string key, got numeric value") + } + value, err := strconv.ParseInt(pretendByteIsString(data), 10, 64) + if err != nil { + return err + } + j.currentOp.AddSignedPair(recID, value) + default: + 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 +} + +// 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 { + switch dataType { + case jsonparser.Number, jsonparser.String: + id, err := j.valueKeys.ID(data) + if err != nil { + return err + } + j.currentOp.AddPair(recID, id) + default: + return fmt.Errorf("expecting integer value, got %v", dataType) + } + 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 { + value := uint64(0) + switch typ { + case jsonparser.String: + if string(data) != "true" && string(data) != "false" { + return fmt.Errorf("expecting boolean, got %q", data) + } + // if it's exactly "true" or "false" let's be forgiving + fallthrough + case jsonparser.Boolean: + if data[0] == 't' { + value = 1 + } + case jsonparser.Number: + v, err := jsonparser.GetInt(data) + if err != nil { + return err + } + if v == 1 { + value = 1 + } else if v != 0 { + return errors.New("boolean should be true/false/0/1") + } + default: + return errors.New("boolean should be true/false/0/1") + } + j.currentOp.AddPair(recID, value) + 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 { + 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) { + switch string(key) { + case "time": + switch dataType { + case jsonparser.String: + stamp, err = time.Parse(time.RFC3339, pretendByteIsString(value)) + case jsonparser.Number: + var unix int64 + unix, err = strconv.ParseInt(pretendByteIsString(value), 10, 64) + stamp = time.Unix(unix, 0) + default: + return fmt.Errorf("expecting time, got %q", value) + } + case "values": + err = j.decodeSetOrValue(dataType, value, func(id uint64) error { + j.scratch = append(j.scratch, id) + return nil + }) + } + return err + }) + if err != nil { + return err + } + if len(j.scratch) > 0 { + defer func() { + // mark these as consumed so if we get called + // again, and don't see a "values" key, we don't reuse them. + j.scratch = j.scratch[:0] + }() + unix := stamp.UnixNano() + for _, value := range j.scratch { + j.currentOp.AddStampedPair(recID, value, unix) + } + } + 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) { + var stamp time.Time + switch dataType { + case jsonparser.String: + stamp, err = time.Parse(time.RFC3339Nano, pretendByteIsString(data)) + if err != nil { + return fmt.Errorf("parsing timestamp: %w", err) + } + j.currentOp.AddSignedPair(recID, (stamp.UnixNano()/j.scaleUnit)-j.epoch) + case jsonparser.Number: + // We could in theory convert this to a time, then convert it + // back, by multiplying by scaleUnit, then dividing. Or... not. + i64, err := strconv.ParseInt(pretendByteIsString(data), 10, 64) + if err != nil { + return fmt.Errorf("parsing numeric timestamp: %w", err) + } + j.currentOp.AddSignedPair(recID, i64-j.epoch) + } + 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 { + switch dataType { + case jsonparser.String, jsonparser.Number: + value, err := jsonparser.GetFloat(data) + if err != nil { + return err + } + j.currentOp.AddSignedPair(recID, int64(value*float64(j.scaleUnit))) + default: + return fmt.Errorf("expecting floating-point value, got %v", dataType) + } + + return nil +} + +func (codec *JSONCodec) ParseKeyedRecords(data []byte) (err error) { + return jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error { + id, err := codec.recKeys.ID(key) + if err != nil { + return err + } + if codec.currentOp.OpType == OpWrite { + codec.currentOp.ClearRecordIDs = append(codec.currentOp.ClearRecordIDs, id) + } + return jsonparser.ObjectEach(value, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error { + fieldCodec, ok := codec.fields[string(key)] + if !ok { + return errFieldNotFound{string(key)} + } + err := fieldCodec.decode(id, dataType, value) + if err != nil { + return fmt.Errorf("parsing value for field %q: %v", key, err) + } + return nil + }) + }) +} + +func (codec *JSONCodec) ParseOperation(data []byte) (op *Operation, err error) { + op = &Operation{FieldOps: make(map[string]*FieldOperation, len(codec.fields))} + codec.currentOp = op + err = jsonparser.ObjectEach(data, func(key []byte, value []byte, dataType jsonparser.ValueType, offset int) error { + switch string(key) { + case "action": + op.OpType, err = ParseOpType(string(value)) + if err != nil { + return fmt.Errorf("unknown action type %q", value) + } + case "records": + for name, fieldCodec := range codec.fields { + fieldOp := &FieldOperation{} + op.FieldOps[name] = fieldOp + // cache this op so we don't have to do the lookups every time + fieldCodec.currentOp = fieldOp + } + err = codec.ParseKeyedRecords(value) + if err != nil { + return fmt.Errorf("parsing records: %v", err) + } + case "record_ids": + var id uint64 + var idErr error + _, err = jsonparser.ArrayEach(value, func(value []byte, dataType jsonparser.ValueType, offset int, unused error) { + id, idErr = codec.recKeys.ID(value) + if idErr != nil { + return + } + op.ClearRecordIDs = append(op.ClearRecordIDs, id) + + }) + // if we got an error converting an ID, error out with it here. + // we can't stop the ArrayEach early, though? + if idErr != nil { + return idErr + } + if err != nil { + return err + } + case "fields": + _, err = jsonparser.ArrayEach(value, func(value []byte, dataType jsonparser.ValueType, offset int, unused error) { + op.ClearFields = append(op.ClearFields, string(value)) + }) + if err != nil { + return err + } + default: + return fmt.Errorf("unknown operation field %q", key) + } + return nil + }) + if op.OpType == OpNone { + return nil, fmt.Errorf("action not specified") + } + return op, err +} + +func (codec *JSONCodec) Parse(r io.Reader) (req *Request, err error) { + data, err := ioutil.ReadAll(r) + if err != nil { + return nil, err + } + return codec.ParseBytes(data) +} + +func (codec *JSONCodec) ParseBytes(data []byte) (req *Request, err error) { + var ops []*Operation + var lastErr error + _, err = jsonparser.ArrayEach(data, func(value []byte, dataType jsonparser.ValueType, offset int, err error) { + switch dataType { + case jsonparser.Object: + op, err := codec.ParseOperation(value) + if err != nil { + lastErr = fmt.Errorf("parsing operation: %v", err) + return + } + ops = append(ops, op) + default: + lastErr = fmt.Errorf("expected operation, found %s", dataType) + } + }) + if lastErr != nil { + return nil, lastErr + } + if err != nil { + return nil, err + } + // and now, key translation! + var keyMap []uint64 + if codec.keyLookup != nil { + keyMap, err = MapForStringTable(codec.recKeys, codec.keyLookup) + if err != nil { + return nil, fmt.Errorf("trying to find record key mapping: %w", err) + } + } + 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 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) + } + } + } + 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 { + return nil, fmt.Errorf("mapping record keys for clear op: %w", err) + } + } + op.Sort() + } + // For clear/delete, that's all we need to do; there's no meaningful fieldops under them. + if op.OpType == OpClear || op.OpType == OpDelete { + continue + } + for field, fieldOp := range op.FieldOps { + if keyMap != nil { + if err = fieldOp.TranslateKeys(keyMap); err != nil { + return nil, fmt.Errorf("mapping record keys for op on %q: %w", field, err) + } + } + if fieldTranslate, ok := valueMaps[field]; ok { + if err = fieldTranslate(fieldOp); err != nil { + return nil, fmt.Errorf("mapping values for op on %q: %w", field, err) + } + } + // Sort by column keys, for now. + fieldOp.Sort() + // Write op will also want to clear every field we saw. + if op.OpType == OpWrite { + op.ClearFields = append(op.ClearFields, field) + } + } + } + return &Request{Ops: ops}, nil +} + +type errFieldNotFound struct { + field string +} + +func (err errFieldNotFound) Error() string { + return fmt.Sprintf("field not found: %q", err.field) +} diff --git a/ingest/codec_test.go b/ingest/codec_test.go new file mode 100644 index 000000000..fe3479273 --- /dev/null +++ b/ingest/codec_test.go @@ -0,0 +1,160 @@ +// 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 ( + "testing" + "time" +) + +func unusableSampleTranslator(keys ...string) (map[string]uint64, error) { + out := make(map[string]uint64, len(keys)) + for _, key := range keys { + out[key] = uint64(len(out)) * 13 + } + return out, nil +} + +func TestSimpleCodec(t *testing.T) { + c, _ := NewJSONCodec(nil) + _ = c.AddSetField("set", nil) + _ = c.AddSetField("setKeys", unusableSampleTranslator) + _ = c.AddMutexField("mutex", nil) + _ = c.AddMutexField("mutexKeys", unusableSampleTranslator) + _ = c.AddTimeQuantumField("tq", nil) + _ = c.AddIntField("int", nil) + _ = c.AddIntField("intKeys", unusableSampleTranslator) + epoch, err := time.Parse("2006-01-02", "2020-01-01") + if err != nil { + t.Fatalf("can't parse sample epoch time: %v", err) + } + _ = c.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000) + _ = c.AddDecimalField("dec", 2) + _ = c.AddBoolField("bool") + sampleJson := []byte(` +[ + { + "action": "set", + "records": { + "2": { + "set": [ 2 ], + "tq": { + "time": "2006-01-02T15:04:05.999999999Z", + "values": [ 6 ] + }, + "dec": 1.02, + "int": 3, + "mutex": 4, + "mutexKeys": "key-a", + "intKeys": "key-a" + }, + "1234567": { + "set": [ 3 ], + "bool": true + }, + "1": { + "set": [ 2 ], + "bool": false, + "setKeys": [ + "key-a", + "key-b" + ], + "tq": { "values": [ 3, 4 ] } + } + } + }, + { + "action": "clear", + "record_ids": [ 5, 6, 7 ], + "fields": [ "tq" ] + }, + { + "action": "write", + "records": { + "3": { + "set": [ 2 ], + "mutex": 4, + "mutexKeys": "key-a", + "tq": { + "time": "2006-01-02T15:04:05.999999999Z", + "values": [ 6 ] + }, + "ts": "2020-01-01T00:01:00.000000000Z", + "int": 3, + "intKeys": "key-a" + }, + "4": { + "set": [ 3 ], + "ts": 1577836860000 + }, + "5": { + "set": [ 2 ], + "setKeys": [ "key-a", "key-b" ], + "tq": { "values": [ 3, 4 ] } + } + } + }, + { + "action": "delete", + "record_ids": [ 9 ] + }, + { + "action": "set", + "records": { + "2": { + "mutex": 5, + "mutexKeys": "key-b" + } + } + } +] +`) + req, err := c.ParseBytes(sampleJson) + if err != nil { + t.Fatalf("parsing sample buffer: %v", err) + } + t.Logf("req: %#v", req) + for _, op := range req.Ops { + t.Logf("op: %#v", op) + if len(op.ClearRecordIDs) > 0 { + t.Logf(" clearRecordIDs: %d", op.ClearRecordIDs) + } + if len(op.ClearFields) > 0 { + t.Logf(" clearFields: %s", op.ClearFields) + } + for field, fieldOp := range op.FieldOps { + t.Logf(" field %q: %#v", field, fieldOp) + } + } + sharded, err := req.Shard() + if err != nil { + t.Errorf("sharding err: %v", err) + } + for shard, ops := range sharded.Ops { + t.Logf("shard %d:", shard) + for _, op := range ops { + t.Logf(" op: %q", op.OpType.String()) + if len(op.ClearRecordIDs) > 0 { + t.Logf(" clearRecordIDs: %d", op.ClearRecordIDs) + } + if len(op.ClearFields) > 0 { + t.Logf(" clearFields: %s", op.ClearFields) + } + for field, fieldOp := range op.FieldOps { + t.Logf(" field %q: %#v", field, fieldOp) + } + } + } +} diff --git a/ingest/doc.go b/ingest/doc.go new file mode 100644 index 000000000..c9e79bfe7 --- /dev/null +++ b/ingest/doc.go @@ -0,0 +1,30 @@ +// 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 provides tooling for accepting record-oriented data updates +// and converting them to data that can be efficiently merged into stored +// data. Nia's original description: +// +// but the overall pipeline is: +// 1. fetch the schema and use it to configure the codec +// 2. parse the data with the codec into vectors, while stuffing temp record key mappings into a string table +// 3. call *CreateKeys on the cluster for all of the things +// 4. generate an ID remapping table for record keys and apply it to all of the vectors +// 5. remap the string keys +// 6. group each vector by shard +// 7. convert the shard vectors into matrix updates +// 8. combine those matrix updates into a shard update +// 9. send the shard updates out over the internal client +// 10. the nodes apply them to RBF +package ingest diff --git a/ingest/op.go b/ingest/op.go new file mode 100644 index 000000000..1006de992 --- /dev/null +++ b/ingest/op.go @@ -0,0 +1,322 @@ +// 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" + "sort" + + "github.com/molecula/featurebase/v2/shardwidth" +) + +type OpType uint8 + +const ( + OpNone = OpType(iota) + OpSet + OpRemove + OpClear + OpWrite + OpDelete +) + +var opNames = []string{ + "none", + "set", + "remove", + "clear", + "write", + "delete", +} + +func (o OpType) String() string { + if int(o) < len(opNames) { + return opNames[o] + } + return fmt.Sprintf("invalid-optype-%d", o) +} + +func ParseOpType(s string) (OpType, error) { + for i, v := range opNames[1:] { + if s == v { + return OpType(i + 1), nil + } + } + return 0, fmt.Errorf("unknown operation type %q", s) +} + +// Operation represents a single set of changes to make to +// the stored data, which means some combination of clearing +// columns, clearing individual bits, or setting bits or values. +// The same data structure can be used whether this represents the +// whole database operation or a single shard's values. +// +// Operations can specify individual per-field operations, which +// have maps of record IDs to values. They can also have a set of +// record IDs and fields to clear. A Clear operation will have only +// 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. +type Operation struct { + OpType OpType + ClearRecordIDs []uint64 + ClearFields []string + FieldOps map[string]*FieldOperation +} + +// FieldOperation is the specific set of changes to make to a given +// field. +// +// For a Clear operation, values can be an empty array. For Set or Remove +// operations on sets, RecordIDs can contain duplicates. Times should +// be empty except for time-quantum fields. +type FieldOperation struct { + RecordIDs []uint64 + Values []uint64 + // For int/timestamp/decimal, this is the value + // For time-quantum, this is the timestamp + // No field has both signed values and timestamps. + // This is not a place of honor. + Signed []int64 +} + +var _ sort.Interface = &FieldOperation{} + +// We implement sort.Interface here so we don't have to write sort code. +// This should probably be replaced by smarter sorting later if it's a +// performance issue. +func (f *FieldOperation) Len() int { + return len(f.RecordIDs) +} + +func (f *FieldOperation) Less(i, j int) bool { + return f.RecordIDs[i] < f.RecordIDs[j] +} + +func (f *FieldOperation) Swap(i, j int) { + f.RecordIDs[i], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[i] + if f.Values != nil { + f.Values[i], f.Values[j] = f.Values[j], f.Values[i] + } + if f.Signed != nil { + f.Signed[i], f.Signed[j] = f.Signed[j], f.Signed[i] + } +} + +// Sort sorts the values by record ID. It is not a stable sort. +func (f *FieldOperation) Sort() { + sort.Sort(f) +} + +// Similarly, do this for Operation, which is used only in the Clear case. +var _ sort.Interface = &Operation{} + +// We implement sort.Interface here so we don't have to write sort code. +// This should probably be replaced by smarter sorting later if it's a +// performance issue. +func (o *Operation) Len() int { + return len(o.ClearRecordIDs) +} + +func (o *Operation) Less(i, j int) bool { + return o.ClearRecordIDs[i] < o.ClearRecordIDs[j] +} + +func (o *Operation) Swap(i, j int) { + o.ClearRecordIDs[i], o.ClearRecordIDs[j] = o.ClearRecordIDs[j], o.ClearRecordIDs[i] +} + +// Sort sorts the values by record ID. It is not a stable sort. +func (o *Operation) Sort() { + sort.Sort(o) +} + +type ShardedFieldOperation map[uint64]*FieldOperation + +// Shard() divides the FieldOperation's values up into corresponding chunks +// based on the shards of record IDs. Record IDs should be sorted before this +// happens. +func (f *FieldOperation) Shard() ShardedFieldOperation { + if len(f.RecordIDs) == 0 { + return nil + } + shards, ends := shardwidth.FindShards(f.RecordIDs) + prev := 0 + op := make(ShardedFieldOperation, len(shards)) + 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] + } + op[shard] = subOp + prev = endIndex + } + return op +} + +// AddPair adds a record ID/value pair where the value is unsigned, as +// when used with set/mutex/time quantum fields. +func (f *FieldOperation) AddPair(rec uint64, value uint64) { + f.RecordIDs = append(f.RecordIDs, rec) + f.Values = append(f.Values, value) +} + +// AddSignedPair adds a record ID/value pair where the value is signed, +// as when used with int/decimal/timestamp fields. +func (f *FieldOperation) AddSignedPair(rec uint64, value int64) { + f.RecordIDs = append(f.RecordIDs, rec) + f.Signed = append(f.Signed, value) +} + +// AddStampedPair adds a record/value pair plus a time, which is just +// a Unix time in seconds. (Note, no scaling here; timestamp fields are +// scaled int fields, this is for time quantums.) +func (f *FieldOperation) AddStampedPair(rec uint64, value uint64, stamp int64) { + f.RecordIDs = append(f.RecordIDs, rec) + f.Values = append(f.Values, value) + f.Signed = append(f.Signed, stamp) +} + +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 + Ops []*Operation +} + +// Request is a complete ingest request, which may be any combination +// of operations, which may apply to multiple shards. +type Request struct { + Ops []*Operation +} + +// ShardedRequest is an ingest request, split up into individual per-shard +// operations. +type ShardedRequest struct { + Ops map[uint64][]*Operation +} + +// Shard converts a request into the same request, only sharded. +func (r *Request) Shard() (*ShardedRequest, error) { + if len(r.Ops) == 0 { + return &ShardedRequest{Ops: nil}, nil + } + req := make(map[uint64][]*Operation) + shards := make(map[uint64]*Operation) + // we're getting per-field things, which we want to divide per-shard, + // and return to per-shard sets of per-field things, so we're inverting + // the structure. + for _, op := range r.Ops { + // for clear and write ops, we also need to split up the + // ClearRecords values, which may be distinct from the set of + // records for any given field. For Write ops, we'll then end + // up adding in field values for some fields. + 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{}} + } + } + for field, fieldOp := range op.FieldOps { + sharded := fieldOp.Shard() + for shard, data := range sharded { + shardOp, ok := shards[shard] + if !ok { + 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} + shards[shard] = shardOp + shardOp.FieldOps = map[string]*FieldOperation{field: data} + } else { + shardOp.FieldOps[field] = data + } + } + } + for shard, shardOp := range shards { + req[shard] = append(req[shard], shardOp) + } + for k := range shards { + delete(shards, k) + } + + } + return &ShardedRequest{Ops: req}, nil +} + +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.ClearFields) > 0 { + logf(" clearFields: %s", op.ClearFields) + } + for field, fieldOp := range op.FieldOps { + logf(" field %q: %#v", field, fieldOp) + } + } +} diff --git a/ingest/op_test.go b/ingest/op_test.go new file mode 100644 index 000000000..db857ac5e --- /dev/null +++ b/ingest/op_test.go @@ -0,0 +1,134 @@ +// 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_test + +import ( + "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 +} + +var opShardingTestCases = []opShardingTestCase{ + { + name: "sample", + input: ingest.Request{ + Ops: []*ingest.Operation{ + { + OpType: ingest.OpSet, + FieldOps: map[string]*ingest.FieldOperation{ + "shard0": { + RecordIDs: []uint64{0, 1}, + }, + "shard0-1": { + RecordIDs: []uint64{ + 0, + 1 << shardwidth.Exponent, + }, + }, + "shard1": { + RecordIDs: []uint64{ + 1 << shardwidth.Exponent, + 1<>uint64(shift))].end++ +// } +// group := pairs[j:i] +// +// // Assign indices within the group to the buckets. +// { +// var start uint +// for i := range buckets { +// bucket := &buckets[i] +// bucket.start = start +// bucket.end += start +// start = bucket.end +// } +// } +// +// // Split the group into the buckets. +// for i, b := range buckets { +// // There is no need to update the state of the current bucket - we will never reference it again after this. +// i := uint8(i) +// for j := b.start; j < b.end; j++ { +// // This inner loop may run quite a few times for the first few buckets, but no element will be moved more than twice per byte. +// for uint8(group[j].RecordID>>shift) != i { +// // This pair is in the wrong bucket. +// // Swap it into the correct bucket. +// dstBucket := &buckets[uint8(group[j].RecordID>>shift)] +// k := dstBucket.start +// dstBucket.start++ +// group[j], group[k] = group[k], group[j] +// } +// } +// } +// } +// } +// +// // Split the pairs by shard. +// shards := make(map[uint64][]IDPair) +// for i := 0; i < len(pairs); { +// // Select the shard. +// shard := pairs[i].RecordID >> shardwidth.Exponent +// +// // Find all pairs in the shard. +// j := i +// incr := 1 +// for i+incr < len(pairs) && pairs[i+incr].RecordID>>shardwidth.Exponent == shard { +// i += incr +// incr *= 2 +// } +// for ; incr > 0; incr /= 2 { +// if i+incr < len(pairs) && pairs[i+incr].RecordID>>shardwidth.Exponent == shard { +// i += incr +// } +// } +// // that found us the last thing in this shard, so... +// i++ +// +// // Add the shard, referencing the original slice. +// // This sets the cap so that we dont accidentally overwrite other shards data. +// shards[shard] = pairs[j:i:i] +// } +// +// return shards +// } + +// pairsToZigZag converts a set of record-value pairs to Pilosa's zig-zag format. +// This assumes that all pairs are within the same shard. +// func pairsToZigZag(pairs []IDPair) []uint64 { +// const recordMask = (1 << shardwidth.Exponent) - 1 +// +// dst := make([]uint64, len(pairs)) +// for i, p := range pairs { +// dst[i] = (p.ID << shardwidth.Exponent) | (p.RecordID & recordMask) +// } +// +// return dst +// } + +// radixSort64 sorts the data with radix-sort. +// The "shift" is the highest differing bit position, rounded down to a multiple of 8. +// If there are duplicates, this may change the duplicate count for some values. +// func radixSort64(data []uint64, shift uint) { +// if len(data) < 2 { +// return +// } +// if shift <= 8 { +// // The data falls into a 16-bit span, so the remaining digits can be sorted simultaneously with a bitmask. +// maskSort(data) +// return +// } +// +// // Count the values within each bucket. +// var buckets [256]struct { +// start, end uint +// } +// for _, v := range data { +// buckets[uint8(v>>shift)].end++ +// } +// +// // Assign indices within the group to the buckets. +// { +// var start uint +// for i := range buckets { +// bucket := &buckets[i] +// bucket.start = start +// bucket.end += start +// start = bucket.end +// } +// } +// +// // Split the data into the buckets. +// var start uint +// for i, b := range buckets { +// // Replace misplaced values until the contents of the bucket all have the correct digit. +// i := uint8(i) +// for j := b.start; j < b.end; j++ { +// // This inner loop may run quite a few times for the first few buckets, but it will never run more than once-per-element-per-byte. +// for uint8(data[j]>>shift) != i { +// // This pair is in the wrong bucket. +// // Swap it into the correct bucket. +// dstBucket := &buckets[uint8(data[j]>>shift)] +// k := dstBucket.start +// dstBucket.start++ +// data[j], data[k] = data[k], data[j] +// } +// } +// +// // Sort the contents of the bucket. +// data := data[start:b.end] +// switch { +// case len(data) < 64: +// // Use insertion-sort because the data is too small for a more complex algorithm to be efficient. +// for i := 0; i < len(data); i++ { +// for j := i; j > 0 && data[j-1] > data[j]; j-- { +// data[j-1], data[j] = data[j], data[j-1] +// } +// } +// +// default: +// // Sort the next byte recursively. +// radixSort64(data, shift-8) +// } +// start = b.end +// } +// } + +// maskSort sorts integers using a bitmask. +// The values must all fall within one 16-bit span. +// If there are duplicates, this may change the duplicate count for some values. +// func maskSort(data []uint64) { +// if len(data) < 2 { +// return +// } +// +// base := data[0] &^ ((1 << 16) - 1) +// +// // Dump everything into the mask. +// var mask [(1 << 16) / 64]uint64 +// for _, v := range data { +// mask[uint16(v)/64] |= 1 << (v % 64) +// } +// +// // Scan through the set bits in the mask. +// k := 0 +// for i, w := range mask { +// for w != 0 { +// j := bits.TrailingZeros64(w) +// w &^= 1 << j +// data[k] = 64*uint64(i) + uint64(j) + base +// k++ +// } +// } +// +// if k < len(data) { +// // Copy the ending value to fill up the rest of the space. +// // This happens once for each duplicate value. +// endVal := data[k-1] +// for k < len(data) { +// data[k] = endVal +// k++ +// } +// } +// } diff --git a/ingest/sort_test.go b/ingest/sort_test.go new file mode 100644 index 000000000..231ff61e8 --- /dev/null +++ b/ingest/sort_test.go @@ -0,0 +1,184 @@ +// 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 + +// func BenchmarkSort64(b *testing.B) { +// gen := func(n, width uint64) func() []uint64 { +// var data []uint64 +// var once sync.Once +// return func() []uint64 { +// once.Do(func() { +// data = make([]uint64, n) +// var rng rand.PCGSource +// rng.Seed(9001) +// for i := range data { +// data[i] = rng.Uint64() % width +// } +// }) +// +// return data +// } +// } +// +// algos := []struct { +// name string +// maxn uint64 +// fn func([]uint64) []uint64 +// }{ +// { +// name: "stdlib", +// maxn: 1024 * 1024 * 1024, +// fn: stdSort, +// }, +// { +// name: "heap", +// maxn: 1024 * 1024 * 1024, +// fn: heapSort, +// }, +// { +// name: "radix-insertion-mask", +// maxn: 1024 * 1024 * 1024, +// fn: dedupSort64, +// }, +// } +// +// widths := []struct { +// name string +// width uint64 +// }{ +// {"64", 64}, +// {"1K", 1024}, +// {"64K", 64 * 1024}, +// {"1M", 1024 * 1024}, +// {"16M", 16 * 1024 * 1024}, +// {"128M", 128 * 1024 * 1024}, +// {"1B", 1024 * 1024 * 1024}, +// } +// +// counts := []struct { +// name string +// n uint64 +// }{ +// {"64", 64}, +// {"1K", 1024}, +// {"64K", 64 * 1024}, +// {"1M", 1024 * 1024}, +// {"4M", 4 * 1024 * 1024}, +// {"16M", 16 * 1024 * 1024}, +// {"64M", 64 * 1024 * 1024}, +// {"256M", 256 * 1024 * 1024}, +// } +// +// for _, width := range widths { +// width := width +// b.Run(width.name, func(b *testing.B) { +// for _, count := range counts { +// if count.n > width.width { +// continue +// } +// +// count := count +// b.Run(count.name, func(b *testing.B) { +// datasrc := gen(count.n, width.width) +// for _, alg := range algos { +// if count.n > alg.maxn { +// continue +// } +// +// alg := alg +// b.Run(alg.name, func(b *testing.B) { +// data := datasrc() +// buf := make([]uint64, len(data)) +// b.SetBytes(8 * int64(len(buf))) +// +// b.StopTimer() +// b.ResetTimer() +// +// for i := 0; i < b.N; i++ { +// copy(buf, data) +// b.StartTimer() +// alg.fn(buf) +// b.StopTimer() +// } +// }) +// } +// }) +// } +// }) +// } +// } +// +// func heapSort(data []uint64) []uint64 { +// for i, v := range data { +// for i > 0 && v > data[(i-1)/2] { +// data[i] = data[(i-1)/2] +// i = (i - 1) / 2 +// } +// data[i] = v +// } +// { +// heap := data +// for len(heap) > 1 { +// heap[0], heap[len(heap)-1] = heap[len(heap)-1], heap[0] +// heap = heap[:len(heap)-1] +// i := 0 +// for { +// max := i +// if r := 2*i + 1; r < len(heap) && heap[r] > heap[max] { +// max = r +// } +// if l := 2*i + 2; l < len(heap) && heap[l] > heap[max] { +// max = l +// } +// if max == i { +// break +// } +// +// heap[max], heap[i] = heap[i], heap[max] +// i = max +// } +// } +// } +// +// j := 1 +// prev := data[0] +// for _, v := range data[1:] { +// if v == prev { +// continue +// } +// +// data[j] = v +// prev = v +// } +// +// return data[:j] +// } +// +// func stdSort(data []uint64) []uint64 { +// sort.Slice(data, func(i, j int) bool { return data[i] < data[j] }) +// +// j := 1 +// prev := data[0] +// for _, v := range data[1:] { +// if v == prev { +// continue +// } +// +// data[j] = v +// prev = v +// } +// +// return data[:j] +// } diff --git a/ingest/translate.go b/ingest/translate.go new file mode 100644 index 000000000..7ce7ea6dd --- /dev/null +++ b/ingest/translate.go @@ -0,0 +1,15 @@ +// 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 diff --git a/ingest/update.go b/ingest/update.go new file mode 100644 index 000000000..8101bb683 --- /dev/null +++ b/ingest/update.go @@ -0,0 +1,256 @@ +// 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 ( + "github.com/molecula/featurebase/v2/roaring" +) + +// ShardUpdate is an update request for a shard. +type ShardUpdate struct { + // TODO: include schema version + + // Sets are the updates to set fields. + Sets map[string]SetMatrixUpdate + + // Mutexes are the updates to mutex fields. + Mutexes map[string]MutexMatrixUpdate + + // TimeTensors are the updates to time fields. + TimeTensors map[string]TimeTensorUpdate + + // Ints are the updates to int fields. + Ints map[string]IntUpdate +} + +// Convert the ID vector to a raw update that can be imported. +// All records are assumed to fall within a shard. +// func (vec IDVector) Convert() (MutexMatrixUpdate, error) { +// // Before converting the vector, do a sanity-check for duplicates. +// dedup := make(map[uint64]struct{}, len(vec.Updates)+len(vec.Clears)) +// for _, p := range vec.Updates { +// if _, ok := dedup[p.RecordID]; !ok { +// return MutexMatrixUpdate{}, errors.New("input contains conflicting updates") +// } +// +// dedup[p.RecordID] = struct{}{} +// } +// for _, p := range vec.Clears { +// if _, ok := dedup[p]; !ok { +// return MutexMatrixUpdate{}, errors.New("input contains conflicting updates") +// } +// +// dedup[p] = struct{}{} +// } +// +// // Convert the updates to a bitmap. +// updates, err := idPairsToBitmap(vec.Updates, false) +// if err != nil { +// return MutexMatrixUpdate{}, errors.Wrap(err, "encoding mutex updates") +// } +// +// // Convert the clears to a bitmap. +// clears, err := idListToBitmap(vec.Clears, false) +// if err != nil { +// return MutexMatrixUpdate{}, errors.Wrap(err, "encoding mutex clears") +// } +// +// // Realign clears bitmap to start of shard. +// if min, ok := clears.Min(); ok { +// min &^= (1 << shardwidth.Exponent) - 1 +// clears = clears.OffsetRange(-min, 0, 1< 0 { +// return SetMatrixUpdate{}, errors.New("set adds and removes overlap") +// } +// +// return SetMatrixUpdate{ +// Add: adds, +// Remove: removes, +// Clear: clears, +// }, nil +// } + +// SetMatrixUpdate is an encoded update request for a set-type (set/mutex) view. +type SetMatrixUpdate struct { + // Add is a bitmap to union the matrix against. + Add *roaring.Bitmap + + // Remove is a bitmap to difference out of the matrix. + // No values will be present in both add and remove. + // This bitmap will not include any records in the clear bitmap. + Remove *roaring.Bitmap + + // Clear is a set of records for which all values should be removed. + // This is applied before add operations. + Clear *roaring.Bitmap +} + +// func idPairsToBitmap(pairs []IDPair, allowDup bool) (*roaring.Bitmap, error) { +// return idListToBitmap(pairsToZigZag(pairs), allowDup) +// } +// +// func idListToBitmap(ids []uint64, allowDup bool) (*roaring.Bitmap, error) { +// vals := dedupSort64(ids) +// if len(vals) != len(ids) && !allowDup { +// return nil, errors.New("input contains duplicate values") +// } +// +// // TODO: make the roaring package actually handle this well +// return roaring.NewBitmap(vals...), nil +// } + +// Convert the int vector to a raw update that can be imported. +// All records are assumed to fall within a shard. +// func (vec IntVector) Convert() (IntUpdate, error) { +// // Convert the clears to a bitmap. +// clears, err := idListToBitmap(vec.Clears, false) +// if err != nil { +// return IntUpdate{}, errors.Wrap(err, "encoding int clears") +// } +// +// // Realign clears bitmap to start of shard. +// if min, ok := clears.Min(); ok { +// min &^= (1 << shardwidth.Exponent) - 1 +// clears = clears.OffsetRange(-min, 0, 1< 0 { +// return IntUpdate{}, errors.New("update duplicated with a clear") +// } +// +// return IntUpdate{ +// BSI: updates, +// Clear: clears, +// }, nil +// } + +// IntUpdate is an encoded update request for a BSI (int/timestamp/etc.) view. +type IntUpdate struct { + // BSI is a bitmap of new BSI data to overwrite existing values. + BSI *roaring.Bitmap + + // Clear is a set of records to assign null. + Clear *roaring.Bitmap +} diff --git a/ingest/vec.go b/ingest/vec.go new file mode 100644 index 000000000..082ff8575 --- /dev/null +++ b/ingest/vec.go @@ -0,0 +1,124 @@ +// 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" + "reflect" + "strconv" + "unsafe" + + "github.com/pkg/errors" +) + +// StringTable is a mapping of strings to temporary IDs. +// All mapped-to IDs fall in the range [0, len). The zero value, +// a nil map, instead just parses the numbers. +// +// We keep the array of names in creation order because we want reproducibility; +// the first key we see is always key 0. Otherwise, the keys are created in +// arbitrary orders. +type StringTable struct { + names []string + values map[string]uint64 +} + +// NewStringTable just creates a string table with a non-nil map. +func NewStringTable() *StringTable { + return &StringTable{values: map[string]uint64{}} +} + +// unsafe is +func pretendByteIsString(data []byte) (result string) { + dH := (*reflect.SliceHeader)(unsafe.Pointer(&data)) + sH := (*reflect.StringHeader)(unsafe.Pointer(&result)) + sH.Data = dH.Data + sH.Len = dH.Len + return result +} + +// ID returns an ID associated to the string, adding it to the table if it is not already present, +// or parsing an integer if there's no table. +func (tbl *StringTable) ID(in []byte) (uint64, error) { + str := pretendByteIsString(in) + if tbl != nil { + id, ok := tbl.values[str] + if !ok { + id = uint64(len(tbl.values)) + tbl.values[str] = id + tbl.names = append(tbl.names, str) + } + return id, nil + } + return strconv.ParseUint(str, 10, 64) +} + +// SignedID returns an ID associated to the string, adding it to the table if it is not already present, +// or parsing an integer if there's no table. It yields signed values only. +func (tbl *StringTable) IntID(in []byte) (int64, error) { + str := pretendByteIsString(in) + if tbl != nil { + id, ok := tbl.values[str] + if !ok { + id = uint64(len(tbl.values)) + tbl.values[str] = id + tbl.names = append(tbl.names, str) + } + return int64(id), nil + } + return strconv.ParseInt(str, 10, 64) +} + +// MapForStringTable, given a string table mapping strings to consecutive +// 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...) + if err != nil { + return nil, err + } + if len(lookedUp) != len(tbl.names) { + return nil, fmt.Errorf("missing keys: expected %d keys, got %d", len(tbl.values), len(lookedUp)) + } + out := make([]uint64, len(tbl.names)) + for i, v := range tbl.names { + out[i] = lookedUp[v] + } + 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)) + } +} + +// TODO: bool + +// TODO: timestamp (just sugar on top of IntVector) diff --git a/roaring/unmarshal_binary.go b/roaring/unmarshal_binary.go index a29bba516..7d8061a99 100644 --- a/roaring/unmarshal_binary.go +++ b/roaring/unmarshal_binary.go @@ -15,6 +15,7 @@ package roaring import ( + "bytes" "errors" "io" "unsafe" @@ -104,6 +105,16 @@ func (b *Bitmap) UnmarshalBinary(data []byte) (err error) { return nil } +func (b *Bitmap) MarshalBinary() ([]byte, error) { + var buf bytes.Buffer + _, err := b.WriteTo(&buf) + if err != nil { + return nil, err + } + + return buf.Bytes(), nil +} + // InspectBinary reads a roaring bitmap, plus a possible ops log, // and reports back on the contents, including distinguishing between // the original ops log and the post-ops-log contents. diff --git a/time.go b/time.go index a17d37726..4253a4d1e 100644 --- a/time.go +++ b/time.go @@ -42,6 +42,13 @@ func (q TimeQuantum) HasDay() bool { return strings.ContainsRune(string(q), 'D') // HasHour returns true if the quantum contains a 'H' unit. func (q TimeQuantum) HasHour() bool { return strings.ContainsRune(string(q), 'H') } +func (q TimeQuantum) Granularity() rune { + var g rune + for _, g = range q { + } + return g +} + // Valid returns true if q is a valid time quantum value. func (q TimeQuantum) Valid() bool { switch q {