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)