From 016765d8a231f844e3b30c1bfdbb09281564ab32 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 10 Aug 2021 13:05:30 -0500 Subject: [PATCH] Prototype ingest API This partially-implemented prototype of the ingest API is based on our programmatic ingest API reference. It has noticable limitations, most crucially that it doesn't handle multi-node clusters right now. However, it basically implements the expected semantics. There's some noticeable performance issues to do with the high overhead of sorting bits in order to import them efficiently, but this is fixable. We also add the hooks to the internal client, and make the finisher logic a bit smarter. Much of this code was originally by Nia Weiss, but it's been merged and restructured a bit to get things broken into logical commits. --- api.go | 193 +++++++++++++ api_test.go | 154 ++++++++++ field.go | 23 ++ fragment.go | 28 +- go.mod | 1 + go.sum | 2 + http/client.go | 55 ++++ ingest/codec.go | 561 ++++++++++++++++++++++++++++++++++++ ingest/codec_test.go | 160 ++++++++++ ingest/doc.go | 30 ++ ingest/op.go | 322 +++++++++++++++++++++ ingest/op_test.go | 134 +++++++++ ingest/shard.go | 15 + ingest/sort.go | 245 ++++++++++++++++ ingest/sort_test.go | 184 ++++++++++++ ingest/translate.go | 15 + ingest/update.go | 256 ++++++++++++++++ ingest/vec.go | 124 ++++++++ roaring/unmarshal_binary.go | 11 + time.go | 7 + 20 files changed, 2519 insertions(+), 1 deletion(-) create mode 100644 ingest/codec.go create mode 100644 ingest/codec_test.go create mode 100644 ingest/doc.go create mode 100644 ingest/op.go create mode 100644 ingest/op_test.go create mode 100644 ingest/shard.go create mode 100644 ingest/sort.go create mode 100644 ingest/sort_test.go create mode 100644 ingest/translate.go create mode 100644 ingest/update.go create mode 100644 ingest/vec.go 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 {