From f019cc740923af39eb9d78b6ba42031cfcf48196 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 26 Jul 2021 13:29:55 -0500 Subject: [PATCH] make import correctly reflect that it needs a single shard always In fact, we have a number of things assuming that values passed to Import always fit within a single known shard, so, drop all the extra complexity around this, drop the computation of fancy view/shard keys, and so on. There's a lot of room left to improve this probably but it's at least better, I think. Unfortunately, there's a handful of things, basically all of which are test cases, which were relying on this, so, we also add functionality for splitting import requests by shards. But this allows us to stop duplicating each shard's inputs one at a time... which turns out to mean that we now care that the import operation can write back to the import request. This only affects test cases, so we adopt a crufty hack involving cloning import requests in those rare cases, and also when reusing the same column IDs to write to the existence field that we'd be using later to write to another field. Note that even if we weren't overwriting the column IDs with positions, we'd be sorting the column/row ID lists by row-then-column, which means we'd still be corrupting the column ID lists. This may want to change at some point. We also reuse a single Tx for all the views, because DB-per-shard means that should work fine, and reduces the cost of doing these updates, probably. --- api.go | 27 +++++---- api_test.go | 4 +- executor_test.go | 11 +++- field.go | 121 ++++++++++++++++++++++++-------------- field_internal_test.go | 11 ++-- fragment_internal_test.go | 4 ++ handler.go | 118 +++++++++++++++++++++++++++++++++++++ http/client.go | 2 +- index.go | 5 -- tx_test.go | 4 +- 10 files changed, 236 insertions(+), 71 deletions(-) diff --git a/api.go b/api.go index 0353b0f42..bbdb8c6b4 100644 --- a/api.go +++ b/api.go @@ -1619,7 +1619,7 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, // Note: req.Shard may not be the only shard imported into here, // so don't expect it to be invariant. if !options.Clear { - if err := importExistenceColumns(qcx, idx, req.ColumnIDs); err != nil { + if err := importExistenceColumns(qcx, idx, req.ColumnIDs, req.Shard); err != nil { api.server.logger.Errorf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) return err } @@ -1629,7 +1629,7 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, } // Import into fragment. - err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, opts...) + err = field.Import(qcx, req.RowIDs, req.ColumnIDs, timestamps, req.Shard, opts...) if err != nil { api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) return errors.Wrap(err, "importing") @@ -1728,8 +1728,9 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu // if we're importing into a specific shard if req.Shard != math.MaxUint64 { // Check that column IDs match the stated shard. - if s1, s2 := req.ColumnIDs[0]/ShardWidth, req.ColumnIDs[len(req.ColumnIDs)-1]/ShardWidth; s1 != s2 && s2 != req.Shard { - return errors.Errorf("shard %d specified, but import spans shards %d to %d", req.Shard, s1, s2) + shard := req.ColumnIDs[0] / ShardWidth + if s2 := req.ColumnIDs[len(req.ColumnIDs)-1] / ShardWidth; (shard != s2) || (shard != req.Shard) { + return errors.Errorf("shard %d specified, but import spans shards %d to %d", req.Shard, shard, s2) } // Validate shard ownership. TODO - we should forward to the // correct node rather than barfing here. @@ -1738,7 +1739,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu } // Import columnIDs into existence field. if !options.Clear { - if err := importExistenceColumns(qcx, idx, req.ColumnIDs); err != nil { + if err := importExistenceColumns(qcx, idx, req.ColumnIDs, shard); err != nil { api.server.logger.Errorf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) return errors.Wrap(err, "importing existence columns") } @@ -1746,17 +1747,17 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu // Import into fragment. if len(req.Values) > 0 { - err = field.importValue(qcx, req.ColumnIDs, req.Values, options) + err = field.importValue(qcx, req.ColumnIDs, req.Values, shard, options) if err != nil { api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) } } else if len(req.TimestampValues) > 0 { - err = field.importTimestampValue(qcx, req.ColumnIDs, req.TimestampValues, options) + err = field.importTimestampValue(qcx, req.ColumnIDs, req.TimestampValues, shard, options) if err != nil { api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) } } else if len(req.FloatValues) > 0 { - err = field.importFloatValue(qcx, req.ColumnIDs, req.FloatValues, options) + err = field.importFloatValue(qcx, req.ColumnIDs, req.FloatValues, shard, options) if err != nil { api.server.logger.Errorf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) } @@ -1814,14 +1815,20 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu return nil } -func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64) error { +func importExistenceColumns(qcx *Qcx, index *Index, columnIDs []uint64, shard uint64) error { ef := index.existenceField() if ef == nil { return nil } existenceRowIDs := make([]uint64, len(columnIDs)) - return ef.Import(qcx, existenceRowIDs, columnIDs, nil) + // 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) } // ShardDistribution returns an object representing the distribution of shards diff --git a/api_test.go b/api_test.go index 4a4418ed9..201a7c89a 100644 --- a/api_test.go +++ b/api_test.go @@ -482,10 +482,10 @@ func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) { } qcx := m0api.Txf().NewQcx() - if err := m0api.Import(ctx, qcx, ir0); err != nil { + if err := m0api.Import(ctx, qcx, ir0.Clone()); err != nil { t.Fatal(err) } - if err := m0api.ImportValue(ctx, qcx, ivr0); err != nil { + if err := m0api.ImportValue(ctx, qcx, ivr0.Clone()); err != nil { t.Fatal(err) } PanicOn(qcx.Finish()) diff --git a/executor_test.go b/executor_test.go index 46b3e00ef..8440adb9d 100644 --- a/executor_test.go +++ b/executor_test.go @@ -4131,10 +4131,17 @@ func TestExecutor_Execute_All(t *testing.T) { req.ColumnIDs[bitCount-1] = uint64((3 * ShardWidth) + 2) m0 := c.GetNode(0) + // the request gets altered by the Import operation now... + reqs, err := req.Clone().ShardSplit() + if err != nil { + t.Fatalf("splitting request into shards: %v", err) + } qcx := m0.API.Txf().NewQcx() - if err := m0.API.Import(context.Background(), qcx, req); err != nil { - t.Fatal(err) + for _, r := range reqs { + if err := m0.API.Import(context.Background(), qcx, r); err != nil { + t.Fatal(err) + } } PanicOn(qcx.Finish()) diff --git a/field.go b/field.go index bfffc3509..dee36745d 100644 --- a/field.go +++ b/field.go @@ -1437,7 +1437,7 @@ func (f *Field) Range(qcx *Qcx, name string, op pql.Token, predicate int64) (*Ro } // Import bulk imports data. -func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, opts ...ImportOption) (err0 error) { +func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, shard uint64, opts ...ImportOption) (err0 error) { // Set up import options. options := &ImportOptions{} @@ -1456,12 +1456,57 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, } else if options.Clear { return errors.New("import clear is not supported with timestamps") } + } else { + // short path: if we don't have any timestamps, we only need + // to write to exactly one view, which is always viewStandard, + // and *every* bit goes into that view, and we already verified that + // everything is in the same shard, so we can skip most of this. + fieldType := f.Type() + if fieldType == FieldTypeBool { + for _, rowID := range rowIDs { + if rowID > 1 { + return errors.New("bool field imports only support values 0 and 1") + } + } + } + tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: f.idx, Shard: shard}) + if err != nil { + return errors.Wrap(err, "qcx.GetTx") + } + var err1 error + defer finisher(&err1) + view, err := f.createViewIfNotExists(viewStandard) + if err != nil { + return errors.Wrapf(err, "creating view %s", viewStandard) + } + + frag, err := view.CreateFragmentIfNotExists(shard) + if err != nil { + return errors.Wrap(err, "creating fragment") + } + + err1 = frag.bulkImport(tx, rowIDs, columnIDs, options) + return err1 } fieldType := f.Type() // Split import data by fragment. - dataByFragment := make(map[importKey]importData) + views := make(map[string]int) + var allData []importData + see := func(name string, columnID uint64, rowID uint64) { + var ok bool + var idx int + if idx, ok = views[name]; !ok { + allData = append(allData, importData{}) + idx = len(allData) + views[name] = idx + } + data := allData[idx] + data.RowIDs = append(data.RowIDs, rowID) + data.ColumnIDs = append(data.ColumnIDs, columnID) + allData[idx] = data + } for i := range rowIDs { rowID, columnID := rowIDs[i], columnIDs[i] @@ -1472,53 +1517,41 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, hasTime := len(timestamps) > i && timestamps[i] != 0 - var standard []string - if !hasTime { - standard = []string{viewStandard} - } else { - // Yes, we mean `0, ts`; ts is an int64 in UnixNano units, - // time.Unix takes seconds-and-nanoseconds. - standard = viewsByTime(viewStandard, time.Unix(0, timestamps[i]).UTC(), q) - if !f.options.NoStandardView { - // In order to match the logic of `SetBit()`, we want bits - // with timestamps to write to both time and standard views. - standard = append(standard, viewStandard) + // attach bit to standard view unless we have a timestamp and + // have the NoStandardView option set + if !hasTime || !f.options.NoStandardView { + see(viewStandard, columnID, rowID) + } + if hasTime { + // attach bit to all the views for this timestamp + views := viewsByTime(viewStandard, time.Unix(0, timestamps[i]).UTC(), q) + for _, view := range views { + see(view, columnID, rowID) } } - - // Attach bit to each standard view. - for _, name := range standard { - key := importKey{View: name, Shard: columnID / ShardWidth} - data := dataByFragment[key] - data.RowIDs = append(data.RowIDs, rowID) - data.ColumnIDs = append(data.ColumnIDs, columnID) - dataByFragment[key] = data - } } - - // Import into each fragment. - for key, data := range dataByFragment { - view, err := f.createViewIfNotExists(key.View) + tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: f.idx, Shard: shard}) + if err != nil { + return errors.Wrap(err, "qcx.GetTx") + } + var err1 error + defer finisher(&err1) + for viewName, idx := range views { + data := allData[idx] + view, err := f.createViewIfNotExists(viewName) if err != nil { - return errors.Wrap(err, "creating view") + return errors.Wrapf(err, "creating view %s", viewName) } - frag, err := view.CreateFragmentIfNotExists(key.Shard) + frag, err := view.CreateFragmentIfNotExists(shard) if err != nil { return errors.Wrap(err, "creating fragment") } - tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: frag.idx, Fragment: frag, Shard: frag.shard}) - if err != nil { - return errors.Wrap(err, "qcx.GetTx") - } - - err1 := frag.bulkImport(tx, data.RowIDs, data.ColumnIDs, options) + err1 = frag.bulkImport(tx, data.RowIDs, data.ColumnIDs, options) if err1 != nil { - finisher(&err1) return err1 } - finisher(nil) } return nil } @@ -1526,7 +1559,7 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []int64, // importFloatValue imports floating point values. In current usage, this // should only ever be called with data for a single shard; the API calls // around this are splitting it up per shard. -func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, options *ImportOptions) error { +func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, shard uint64, options *ImportOptions) error { // convert values to int64 values based on scale ivalues := make([]int64, len(values)) bsig := f.bsiGroup(f.name) @@ -1538,13 +1571,13 @@ func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, ivalues[i] = int64(fval * mult) } // then call importValue - return f.importValue(qcx, columnIDs, ivalues, options) + return f.importValue(qcx, columnIDs, ivalues, shard, options) } // importFloatValue imports timestamp values. In current usage, this // should only ever be called with data for a single shard; the API calls // around this are splitting it up per shard. -func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time.Time, options *ImportOptions) error { +func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time.Time, shard uint64, options *ImportOptions) error { ivalues := make([]int64, len(values)) bsig := f.bsiGroup(f.name) if bsig == nil { @@ -1554,13 +1587,13 @@ func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time for i, t := range values { ivalues[i] = t.UnixNano() / TimeUnitNanos(f.options.TimeUnit) } - return f.importValue(qcx, columnIDs, ivalues, options) + return f.importValue(qcx, columnIDs, ivalues, shard, options) } // importValue bulk imports range-encoded value data. This function should // only be called with data for a single shard; the API calls that wrap // this handle splitting the data up per-shard. -func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, options *ImportOptions) (err0 error) { +func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard uint64, options *ImportOptions) (err0 error) { // no data to import if len(columnIDs) == 0 { return nil @@ -1613,9 +1646,9 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option } f.mu.Unlock() - // Since all data should be for the same shard, we can just compute - // this from the first value. - shard := columnIDs[0] / ShardWidth + if columnIDs[0]/ShardWidth != shard { + return fmt.Errorf("requested import for shard %d, got record ID for shard %d", shard, columnIDs[0]/ShardWidth) + } view, err := f.createViewIfNotExists(viewName) if err != nil { diff --git a/field_internal_test.go b/field_internal_test.go index 41f288016..fc2296845 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -494,7 +494,7 @@ func TestBSIGroup_importValue(t *testing.T) { []uint64{100}, }, } { - if err := f.importValue(qcx, tt.columnIDs, tt.values, options); err != nil { + if err := f.importValue(qcx, tt.columnIDs, tt.values, 0, options); err != nil { t.Fatalf("test %d, importing values: %s", i, err.Error()) } PanicOn(qcx.Finish()) @@ -514,7 +514,8 @@ func TestBSIGroup_importValue(t *testing.T) { func benchmarkFieldImportValues(b *testing.B, qcx *Qcx, bitDepth uint64, f *TestField, cfunc func(uint64) uint64) { batches := makeBenchmarkImportValueData(b, bitDepth, cfunc) for _, req := range batches { - err := f.importValue(qcx, req.ColumnIDs, req.Values, &ImportOptions{}) + // NOTE: We assume everything's in Shard 0 for now. + err := f.importValue(qcx, req.ColumnIDs, req.Values, 0, &ImportOptions{}) if err != nil { b.Fatalf("error importing values: %s", err) } @@ -591,7 +592,7 @@ func TestIntField_MinMaxForShard(t *testing.T) { }, } { t.Run(test.name+strconv.Itoa(i), func(t *testing.T) { - if err := f.importValue(qcx, test.columnIDs, test.values, options); err != nil { + if err := f.importValue(qcx, test.columnIDs, test.values, 0, options); err != nil { t.Fatalf("test %d, importing values: %s", i, err.Error()) } PanicOn(qcx.Finish()) @@ -751,7 +752,7 @@ func TestDecimalField_MinMaxForShard(t *testing.T) { }, } { t.Run(test.name+strconv.Itoa(i), func(t *testing.T) { - if err := f.importFloatValue(qcx, test.columnIDs, test.values, options); err != nil { + if err := f.importFloatValue(qcx, test.columnIDs, test.values, 0, options); err != nil { t.Fatalf("test %d, importing values: %s", i, err.Error()) } @@ -811,7 +812,7 @@ func TestBSIGroup_TxReopenDB(t *testing.T) { []uint64{100}, }, } { - if err := f.importValue(qcx, tt.columnIDs, tt.values, options); err != nil { + if err := f.importValue(qcx, tt.columnIDs, tt.values, 0, options); err != nil { t.Fatalf("test %d, importing values: %s", i, err.Error()) } PanicOn(qcx.Finish()) diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 7de7b5c1f..3820d25fa 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1020,6 +1020,10 @@ func BenchmarkFragment_SetValue(b *testing.B) { } } +// makeBenchmarkImportValueData produces data that's supposed to be all within +// the same shard; for fragment purposes, implicitly shard 0. This also gets +// used by the field tests, but import requests are supposed to be per-shard, +// so it's important that we generate values only within a given shard. func makeBenchmarkImportValueData(b *testing.B, bitDepth uint64, cfunc func(uint64) uint64) []ImportValueRequest { b.StopTimer() column := uint64(0) diff --git a/handler.go b/handler.go index 49c4b1ff5..a07c3d6c9 100644 --- a/handler.go +++ b/handler.go @@ -18,6 +18,7 @@ import ( "encoding/json" "time" + "github.com/molecula/featurebase/v2/shardwidth" "github.com/molecula/featurebase/v2/tracing" "github.com/pkg/errors" ) @@ -129,6 +130,41 @@ type ImportValueRequest struct { scratch []int // scratch space to allow us to get a stable sort in reasonable time } +func (ivr *ImportValueRequest) Clone() *ImportValueRequest { + newIVR := &ImportValueRequest{} + if ivr == nil { + return newIVR + } + *newIVR = *ivr + // don't copy the internal scratch buffer + newIVR.scratch = nil + if len(ivr.ColumnIDs) > 0 { + newIVR.ColumnIDs = make([]uint64, len(ivr.ColumnIDs)) + copy(newIVR.ColumnIDs, ivr.ColumnIDs) + } + if len(ivr.ColumnKeys) > 0 { + newIVR.ColumnKeys = make([]string, len(ivr.ColumnKeys)) + copy(newIVR.ColumnKeys, ivr.ColumnKeys) + } + if len(ivr.Values) > 0 { + newIVR.Values = make([]int64, len(ivr.Values)) + copy(newIVR.Values, ivr.Values) + } + if len(ivr.FloatValues) > 0 { + newIVR.FloatValues = make([]float64, len(ivr.FloatValues)) + copy(newIVR.FloatValues, ivr.FloatValues) + } + if len(ivr.TimestampValues) > 0 { + newIVR.TimestampValues = make([]time.Time, len(ivr.TimestampValues)) + copy(newIVR.TimestampValues, ivr.TimestampValues) + } + if len(ivr.StringValues) > 0 { + newIVR.StringValues = make([]string, len(ivr.StringValues)) + copy(newIVR.StringValues, ivr.StringValues) + } + return newIVR +} + // AtomicRecord applies all its Ivr and Ivr atomically, in a Tx. // The top level Shard has to agree with Ivr[i].Shard and the Iv[i].Shard // for all i included (in Ivr and Ir). The same goes for the top level Index: all records @@ -142,6 +178,19 @@ type AtomicRecord struct { Ir []*ImportRequest // other field types, e.g. single bit } +func (ar *AtomicRecord) Clone() *AtomicRecord { + newAR := &AtomicRecord{Index: ar.Index, Shard: ar.Shard} + newAR.Ivr = make([]*ImportValueRequest, len(ar.Ivr)) + for i, vr := range ar.Ivr { + newAR.Ivr[i] = vr.Clone() + } + newAR.Ir = make([]*ImportRequest, len(ar.Ir)) + for i, vr := range ar.Ir { + newAR.Ir[i] = vr.Clone() + } + return newAR +} + func (ivr *ImportValueRequest) Len() int { return len(ivr.ColumnIDs) } func (ivr *ImportValueRequest) Less(i, j int) bool { if ivr.ColumnIDs[i] < ivr.ColumnIDs[j] { @@ -225,6 +274,75 @@ type ImportRequest struct { Clear bool } +// Clone allows copying an import request. Normally you wouldn't, but +// some import functions are destructive on their inputs, and if you +// want to *re-use* an import request, you might need this. If you're +// using this outside tx_test, something is probably wrong. +func (ir *ImportRequest) Clone() *ImportRequest { + newIR := &ImportRequest{} + if ir == nil { + return newIR + } + *newIR = *ir + if ir.RowIDs != nil { + newIR.RowIDs = make([]uint64, len(ir.RowIDs)) + copy(newIR.RowIDs, ir.RowIDs) + } + if ir.ColumnIDs != nil { + newIR.ColumnIDs = make([]uint64, len(ir.ColumnIDs)) + copy(newIR.ColumnIDs, ir.ColumnIDs) + } + if ir.RowKeys != nil { + newIR.RowKeys = make([]string, len(ir.RowKeys)) + copy(newIR.RowKeys, ir.RowKeys) + } + if ir.ColumnKeys != nil { + newIR.ColumnKeys = make([]string, len(ir.ColumnKeys)) + copy(newIR.ColumnKeys, ir.ColumnKeys) + } + if ir.Timestamps != nil { + newIR.Timestamps = make([]int64, len(ir.Timestamps)) + copy(newIR.Timestamps, ir.Timestamps) + } + return newIR +} + +// ShardSplit splits the request into a slice of import requests. It requires +// that the original request have all elements sorted, and already have +// column IDs, not column keys. +func (ir *ImportRequest) ShardSplit() ([]*ImportRequest, error) { + if ir == nil { + return nil, nil + } + // fix shard + if len(ir.ColumnIDs) < 2 { + ir.Shard = ir.ColumnIDs[0] >> shardwidth.Exponent + return []*ImportRequest{ir}, nil + } + shards, ends := shardwidth.FindShards(ir.ColumnIDs) + out := make([]*ImportRequest, len(shards)) + prev := 0 + for i, shard := range shards { + next := ends[i] + newIR := &ImportRequest{} + *newIR = *ir + newIR.ColumnIDs = ir.ColumnIDs[prev:next:next] + if ir.RowIDs != nil { + newIR.RowIDs = ir.RowIDs[prev:next:next] + } + if ir.RowKeys != nil { + newIR.RowKeys = ir.RowKeys[prev:next:next] + } + if ir.Timestamps != nil { + newIR.Timestamps = ir.Timestamps[prev:next:next] + } + newIR.Shard = shard + out[i] = newIR + prev = next + } + return out, nil +} + // ValidateWithTimestamp ensures that the payload of the request is valid. func (ir *ImportRequest) ValidateWithTimestamp(indexCreatedAt, fieldCreatedAt int64) error { if (ir.IndexCreatedAt != 0 && ir.IndexCreatedAt != indexCreatedAt) || diff --git a/http/client.go b/http/client.go index a51f54233..ce31012d7 100644 --- a/http/client.go +++ b/http/client.go @@ -30,7 +30,7 @@ import ( "strings" "time" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/encoding/proto" pnet "github.com/molecula/featurebase/v2/net" "github.com/molecula/featurebase/v2/topology" diff --git a/index.go b/index.go index 370a328fb..25b7bd5ee 100644 --- a/index.go +++ b/index.go @@ -840,11 +840,6 @@ type IndexOptions struct { TrackExistence bool `json:"trackExistence"` } -type importKey struct { - View string - Shard uint64 -} - type importData struct { RowIDs []uint64 ColumnIDs []uint64 diff --git a/tx_test.go b/tx_test.go index 756fae6e5..e383a86b1 100644 --- a/tx_test.go +++ b/tx_test.go @@ -203,7 +203,7 @@ func TestAPI_ImportAtomicRecord(t *testing.T) { qcx = m0api.Txf().NewQcx() //vv("just before the SECOND ImportAtomicRecord, qcx is %p, should NOT BE NIL", qcx) - err = m0api.ImportAtomicRecord(ctx, qcx, air, opt) + err = m0api.ImportAtomicRecord(ctx, qcx, air.Clone(), opt) //err = m0api.ImportAtomicRecord(ctx, nil, air, opt) if err != pilosa.ErrAborted { PanicOn(fmt.Sprintf("expected ErrTxnAborted but got err='%#v'", err)) @@ -229,7 +229,7 @@ func TestAPI_ImportAtomicRecord(t *testing.T) { // happy path with no power failure half-way through. qcx = m0api.Txf().NewQcx() - err = m0api.ImportAtomicRecord(ctx, qcx, air) + err = m0api.ImportAtomicRecord(ctx, qcx, air.Clone()) PanicOn(err) if err := qcx.Finish(); err != nil { t.Fatal(err)