diff --git a/api.go b/api.go index c9c998848..7f22a52ed 100644 --- a/api.go +++ b/api.go @@ -1430,16 +1430,6 @@ var ErrAborted = fmt.Errorf("error: update was aborted") func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRecord, opts ...ImportOption) error { - // this is because some of the tests pass nil qcx for convenience. - isLocalQcx := false - if qcx == nil { - isLocalQcx = true - qcx = api.Txf().NewQcx() - defer func() { - qcx.Abort() - }() - } - simPowerLoss := false lossAfter := -1 var opt ImportOptions @@ -1491,11 +1481,6 @@ func (api *API) ImportAtomicRecord(ctx context.Context, qcx *Qcx, req *AtomicRec return errors.Wrap(err, "ImportAtomicRecord ImportWithTx") } } - - // got to the end succesfully, so commit if we made the qcx - if isLocalQcx { - return qcx.Finish() - } return nil } @@ -1521,21 +1506,10 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts . if req.Clear { opts = addClearToImportOptions(opts) } - isLocalQcx := false - if qcx == nil { - isLocalQcx = true - qcx = api.Txf().NewQcx() - defer func() { - qcx.Abort() - }() - } err = api.ImportWithTx(ctx, qcx, req, opts...) if err != nil { return err } - if isLocalQcx { - return qcx.Finish() - } return nil } @@ -1752,14 +1726,6 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu // don't keep that list around since we don't need it anymore req.scratch = nil } - isLocalQcx := false - if qcx == nil { - isLocalQcx = true - qcx = api.Txf().NewQcx() - defer func() { - qcx.Abort() - }() - } // if we're importing into a specific shard if req.Shard != math.MaxUint64 { @@ -1847,9 +1813,6 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu if err != nil { return err } - if isLocalQcx { - return qcx.Finish() - } return nil } diff --git a/executor_test.go b/executor_test.go index 834906864..46b3e00ef 100644 --- a/executor_test.go +++ b/executor_test.go @@ -492,7 +492,7 @@ func TestExecutor(t *testing.T) { Set(6, f=1, 2001-01-01T00:00) Set(7, f=1, 2002-01-01T02:00) Set(8, f=1, %s) - + Set(2, f=1, 1999-12-30T00:00) Set(2, f=1, 2002-02-01T00:00) Set(2, f=10, 2001-01-01T00:00)`, nextDayExclusive.Format("2006-01-02T15:04")) @@ -539,7 +539,7 @@ func TestExecutor(t *testing.T) { Set("five", f=1, 2000-02-01T00:00) Set("six", f=1, 2001-01-01T00:00) Set("seven", f=1, 2002-01-01T02:00) - + Set("two", f=1, 1999-12-30T00:00) Set("two", f=1, 2002-02-01T00:00) Set("two", f=10, 2001-01-01T00:00)` @@ -573,7 +573,7 @@ func TestExecutor(t *testing.T) { Set(5, f="foo", 2000-02-01T00:00) Set(6, f="foo", 2001-01-01T00:00) Set(7, f="foo", 2002-01-01T02:00) - + Set(2, f="foo", 1999-12-30T00:00) Set(2, f="foo", 2002-02-01T00:00) Set(2, f="bar", 2001-01-01T00:00)` @@ -608,7 +608,7 @@ func TestExecutor(t *testing.T) { Set("five", f="foo", 2000-02-01T00:00) Set("six", f="foo", 2001-01-01T00:00) Set("seven", f="foo", 2002-01-01T02:00) - + Set("two", f="foo", 1999-12-30T00:00) Set("two", f="foo", 2002-02-01T00:00) Set("two", f="bar", 2001-01-01T00:00)` @@ -643,7 +643,7 @@ func TestExecutor(t *testing.T) { Set(5, f=1, 2000-02-01T00:00) Set(6, f=1, 2001-01-01T00:00) Set(7, f=1, 2002-01-01T02:00) - + Set(2, f=1, 1999-12-30T00:00) Set(2, f=1, 2002-02-01T00:00) Set(2, f=10, 2001-01-01T00:00)` @@ -678,7 +678,7 @@ func TestExecutor(t *testing.T) { Set(5, f=1, 2000-02-01T00:00) Set(6, f=1, 2001-01-01T00:00) Set(7, f=1, 2002-01-01T02:00) - + Set(2, f=1, 1999-12-30T00:00) Set(2, f=1, 2002-02-01T00:00) Set(2, f=10, 2001-01-01T00:00)` @@ -724,7 +724,7 @@ func TestExecutor(t *testing.T) { Set("five", f=1, 2000-02-01T00:00) Set("six", f=1, 2001-01-01T00:00) Set("seven", f=1, 2002-01-01T02:00) - + Set("two", f=1, 1999-12-30T00:00) Set("two", f=1, 2002-02-01T00:00) Set("two", f=10, 2001-01-01T00:00)` @@ -758,7 +758,7 @@ func TestExecutor(t *testing.T) { Set(5, f="foo", 2000-02-01T00:00) Set(6, f="foo", 2001-01-01T00:00) Set(7, f="foo", 2002-01-01T02:00) - + Set(2, f="foo", 1999-12-30T00:00) Set(2, f="foo", 2002-02-01T00:00) Set(2, f="bar", 2001-01-01T00:00)` @@ -793,7 +793,7 @@ func TestExecutor(t *testing.T) { Set("five", f="foo", 2000-02-01T00:00) Set("six", f="foo", 2001-01-01T00:00) Set("seven", f="foo", 2002-01-01T02:00) - + Set("two", f="foo", 1999-12-30T00:00) Set("two", f="foo", 2002-02-01T00:00) Set("two", f="bar", 2001-01-01T00:00)` @@ -5154,10 +5154,13 @@ func TestExecutor_GroupByStrings(t *testing.T) { }); err != nil { t.Fatalf("importing: %v", err) } + m0 := c.GetNode(0) + qcx := m0.API.Txf().NewQcx() + defer qcx.Abort() var v1, v2, v3, v4, v5, v6, v7, v8, v9, v10 int64 = 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 var nv1, nv2, nv3, nv4 int64 = -1, -2, -3, -4 - if err := c.GetNode(0).API.ImportValue(context.Background(), nil, &pilosa.ImportValueRequest{ + if err := m0.API.ImportValue(context.Background(), qcx, &pilosa.ImportValueRequest{ Index: "istring", Field: "v", Shard: 0, @@ -5167,7 +5170,7 @@ func TestExecutor_GroupByStrings(t *testing.T) { t.Fatalf("importing: %v", err) } - if err := c.GetNode(0).API.ImportValue(context.Background(), nil, &pilosa.ImportValueRequest{ + if err := m0.API.ImportValue(context.Background(), qcx, &pilosa.ImportValueRequest{ Index: "istring", Field: "vv", Shard: 0, @@ -5177,7 +5180,7 @@ func TestExecutor_GroupByStrings(t *testing.T) { t.Fatalf("importing: %v", err) } - if err := c.GetNode(0).API.ImportValue(context.Background(), nil, &pilosa.ImportValueRequest{ + if err := m0.API.ImportValue(context.Background(), qcx, &pilosa.ImportValueRequest{ Index: "istring", Field: "nv", Shard: 0, diff --git a/test/cluster.go b/test/cluster.go index 7ff157094..126245a46 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -23,7 +23,7 @@ import ( "testing" "time" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/api/client" "github.com/molecula/featurebase/v2/disco" "github.com/molecula/featurebase/v2/logger" @@ -213,32 +213,37 @@ func (c *Cluster) ImportBitsWithTimestamp(t testing.TB, index, field string, row if com.API.Node().ID != node.ID { continue } - if len(timestamps) == 0 { - err := com.API.Import(context.Background(), nil, &pilosa.ImportRequest{ - Index: index, - Field: field, - Shard: shard, - RowIDs: rowIDs, - ColumnIDs: colIDs, - }) - if err != nil { - t.Fatalf("importing data: %v", err) - } - } else { - ts := byShardTs[shard] - err := com.API.Import(context.Background(), nil, &pilosa.ImportRequest{ - Index: index, - Field: field, - Shard: shard, - RowIDs: rowIDs, - ColumnIDs: colIDs, - Timestamps: ts, - }) - if err != nil { - t.Fatalf("importing data: %v", err) - } + func() { + qcx := com.API.Txf().NewQcx() + defer qcx.Abort() + if len(timestamps) == 0 { + err := com.API.Import(context.Background(), qcx, &pilosa.ImportRequest{ + Index: index, + Field: field, + Shard: shard, + RowIDs: rowIDs, + ColumnIDs: colIDs, + }) + if err != nil { + t.Fatalf("importing data: %v", err) + } + } else { + ts := byShardTs[shard] + err := com.API.Import(context.Background(), qcx, &pilosa.ImportRequest{ + Index: index, + Field: field, + Shard: shard, + RowIDs: rowIDs, + ColumnIDs: colIDs, + Timestamps: ts, + }) + if err != nil { + t.Fatalf("importing data: %v", err) + } + + } + }() - } } } } @@ -262,7 +267,9 @@ func (c *Cluster) ImportKeyKey(t testing.TB, index, field string, valAndRecKeys importRequest.RowKeys[i] = vk[0] importRequest.ColumnKeys[i] = vk[1] } - err := c.GetPrimary().API.Import(context.Background(), nil, importRequest) + qcx := c.GetPrimary().API.Txf().NewQcx() + defer qcx.Abort() + err := c.GetPrimary().API.Import(context.Background(), qcx, importRequest) if err != nil { t.Fatalf("importing keykey data: %v", err) } @@ -292,7 +299,9 @@ func (c *Cluster) ImportTimeQuantumKey(t testing.TB, index, field string, entrie importRequest.Timestamps[i] = entry.Ts } - err := c.GetPrimary().API.Import(context.Background(), nil, importRequest) + qcx := c.GetPrimary().API.Txf().NewQcx() + defer qcx.Abort() + err := c.GetPrimary().API.Import(context.Background(), qcx, importRequest) if err != nil { t.Fatalf("importing keykey data: %v", err) } @@ -318,7 +327,9 @@ func (c *Cluster) ImportIntKey(t testing.TB, index, field string, pairs []IntKey importRequest.Values[i] = pair.Val importRequest.ColumnKeys[i] = pair.Key } - if err := c.GetPrimary().API.ImportValue(context.Background(), nil, importRequest); err != nil { + qcx := c.GetPrimary().API.Txf().NewQcx() + defer qcx.Abort() + if err := c.GetPrimary().API.ImportValue(context.Background(), qcx, importRequest); err != nil { t.Fatalf("importing IntKey data: %v", err) } } @@ -342,7 +353,9 @@ func (c *Cluster) ImportIntID(t testing.TB, index, field string, pairs []IntID) importRequest.Values[i] = pair.Val importRequest.ColumnIDs[i] = pair.ID } - if err := c.GetPrimary().API.ImportValue(context.Background(), nil, importRequest); err != nil { + qcx := c.GetPrimary().API.Txf().NewQcx() + defer qcx.Abort() + if err := c.GetPrimary().API.ImportValue(context.Background(), qcx, importRequest); err != nil { t.Fatalf("importing IntID data: %v", err) } } @@ -367,7 +380,9 @@ func (c *Cluster) ImportIDKey(t testing.TB, index, field string, pairs []KeyID) importRequest.RowIDs[i] = pair.ID importRequest.ColumnKeys[i] = pair.Key } - err := c.GetPrimary().API.Import(context.Background(), nil, importRequest) + qcx := c.GetPrimary().API.Txf().NewQcx() + defer qcx.Abort() + err := c.GetPrimary().API.Import(context.Background(), qcx, importRequest) if err != nil { t.Fatalf("importing IDKey data: %v", err) } diff --git a/tx_test.go b/tx_test.go index e53a91dbe..756fae6e5 100644 --- a/tx_test.go +++ b/tx_test.go @@ -20,7 +20,7 @@ import ( "strings" "testing" - "github.com/molecula/featurebase/v2" + pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/server" "github.com/molecula/featurebase/v2/storage" @@ -164,7 +164,12 @@ func TestAPI_ImportAtomicRecord(t *testing.T) { //vv("BEFORE the first ImportAtomicRecord!") - if err := m0api.ImportAtomicRecord(ctx, nil, air); err != nil { + qcx := m0api.Txf().NewQcx() + if err := m0api.ImportAtomicRecord(ctx, qcx, air); err != nil { + qcx.Abort() + t.Fatal(err) + } + if err := qcx.Finish(); err != nil { t.Fatal(err) } @@ -196,7 +201,7 @@ func TestAPI_ImportAtomicRecord(t *testing.T) { air = createAIRUpdate(expectedBalEndingAcct0, expectedBalEndingAcct1) - qcx := m0api.Txf().NewQcx() + 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, nil, air, opt) @@ -223,9 +228,12 @@ func TestAPI_ImportAtomicRecord(t *testing.T) { // happy path with no power failure half-way through. - err = m0api.ImportAtomicRecord(ctx, nil, air) + qcx = m0api.Txf().NewQcx() + err = m0api.ImportAtomicRecord(ctx, qcx, air) PanicOn(err) - + if err := qcx.Finish(); err != nil { + t.Fatal(err) + } eb0, eb1 := queryBalances(m0api, acctOwnerID, fieldAcct0, fieldAcct1, index) // should have been applied this time. @@ -240,7 +248,11 @@ func TestAPI_ImportAtomicRecord(t *testing.T) { air.Ivr[1].Clear = true air.Ir[0].Clear = true - err = m0api.ImportAtomicRecord(ctx, nil, air) + qcx = m0api.Txf().NewQcx() + err = m0api.ImportAtomicRecord(ctx, qcx, air) + if err := qcx.Finish(); err != nil { + t.Fatal(err) + } PanicOn(err) eb0, eb1 = queryBalances(m0api, acctOwnerID, fieldAcct0, fieldAcct1, index)