From 6aadd13095b07c2d1ca2cc21f809012e6dcde69b Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Tue, 3 Nov 2020 23:57:46 +0000 Subject: [PATCH] qcx.GetTx returns an error - to indicate that the query context is already done. - handles the case where the import worker is interrupted early by a ctx cancellation, thus avoiding a panic. --- api.go | 5 +- executor.go | 199 ++++++++++++++++++++++++++----------- field.go | 12 ++- txfactory.go | 11 +- txfactory_internal_test.go | 6 +- view.go | 5 +- 6 files changed, 166 insertions(+), 72 deletions(-) diff --git a/api.go b/api.go index e236fc0ab..dcd137f58 100644 --- a/api.go +++ b/api.go @@ -377,7 +377,10 @@ func importWorker(importWork chan importJob) { } if err := func() (err1 error) { - tx, finisher := j.qcx.GetTx(Txo{Write: writable, Index: j.field.idx, Shard: j.shard}) + tx, finisher, err := j.qcx.GetTx(Txo{Write: writable, Index: j.field.idx, Shard: j.shard}) + if err != nil { + return err + } defer finisher(&err1) var doClear bool diff --git a/executor.go b/executor.go index 77d169b65..97c750e74 100644 --- a/executor.go +++ b/executor.go @@ -891,11 +891,14 @@ func (e *executor) executeFieldValueCall(ctx context.Context, qcx *Qcx, index st return other, nil } -func (e *executor) executeFieldValueCallShard(ctx context.Context, qcx *Qcx, field *Field, col uint64, shard uint64) (_ ValCount, err error) { +func (e *executor) executeFieldValueCallShard(ctx context.Context, qcx *Qcx, field *Field, col uint64, shard uint64) (_ ValCount, err0 error) { idx := e.Holder.Index(field.index) - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return ValCount{}, err + } + defer finisher(&err0) value, exists, err := field.Value(tx, col) if err != nil { @@ -1426,10 +1429,13 @@ func (e *executor) executeDistinctShard(ctx context.Context, qcx *Qcx, index str return executeDistinctShardBSI(ctx, qcx, idx, fieldName, shard, bsig, filterBitmap) } -func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldName string, shard uint64, filterBitmap *roaring.Bitmap) (result SignedRow, err error) { +func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldName string, shard uint64, filterBitmap *roaring.Bitmap) (result SignedRow, err0 error) { index := idx.Name() - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return SignedRow{}, err + } + defer finisher(&err0) fragData, _, err := tx.ContainerIterator(index, fieldName, "standard", shard, 0) if err != nil { @@ -1486,14 +1492,17 @@ func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldNam return SignedRow{Pos: NewRowFromBitmap(rows)}, nil } -func executeDistinctShardBSI(ctx context.Context, qcx *Qcx, idx *Index, fieldName string, shard uint64, bsig *bsiGroup, filterBitmap *roaring.Bitmap) (result SignedRow, err error) { +func executeDistinctShardBSI(ctx context.Context, qcx *Qcx, idx *Index, fieldName string, shard uint64, bsig *bsiGroup, filterBitmap *roaring.Bitmap) (result SignedRow, err0 error) { view := viewBSIGroupPrefix + fieldName index := idx.Name() depth := uint64(bsig.BitDepth) offset := bsig.Base - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return SignedRow{}, err + } + defer finisher(&err0) existsBitmap, err := tx.OffsetRange(index, fieldName, view, shard, ShardWidth*shard, ShardWidth*0, ShardWidth*1) if err != nil { @@ -1596,7 +1605,7 @@ func executeDistinctShardBSI(ctx context.Context, qcx *Qcx, idx *Index, fieldNam } // executeSumCountShard calculates the sum and count for bsiGroups on a shard. -func (e *executor) executeSumCountShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, filter *Row, shard uint64) (_ ValCount, err error) { +func (e *executor) executeSumCountShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, filter *Row, shard uint64) (_ ValCount, err0 error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSumCountShard") defer span.Finish() @@ -1631,8 +1640,11 @@ func (e *executor) executeSumCountShard(ctx context.Context, qcx *Qcx, index str return ValCount{}, nil } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: fragment, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: fragment, Shard: shard}) + if err != nil { + return ValCount{}, err + } + defer finisher(&err0) sumspan, _ := tracing.StartSpanFromContext(ctx, "Executor.executeSumCountShard_fragment.sum") defer sumspan.Finish() @@ -1647,7 +1659,7 @@ func (e *executor) executeSumCountShard(ctx context.Context, qcx *Qcx, index str } // executeMinShard calculates the min for bsiGroups on a shard. -func (e *executor) executeMinShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ ValCount, err error) { +func (e *executor) executeMinShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ ValCount, err0 error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinShard") defer span.Finish() @@ -1669,13 +1681,16 @@ func (e *executor) executeMinShard(ctx context.Context, qcx *Qcx, index string, return ValCount{}, nil } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return ValCount{}, err + } + defer finisher(&err0) return field.MinForShard(tx, shard, filter) } // executeMaxShard calculates the max for bsiGroups on a shard. -func (e *executor) executeMaxShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ ValCount, err error) { +func (e *executor) executeMaxShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ ValCount, err0 error) { idx := e.Holder.Index(index) @@ -1695,13 +1710,17 @@ func (e *executor) executeMaxShard(ctx context.Context, qcx *Qcx, index string, return ValCount{}, nil } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return ValCount{}, err + } + + defer finisher(&err0) return field.MaxForShard(tx, shard, filter) } // executeMinRowShard returns the minimum row ID for a shard. -func (e *executor) executeMinRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ PairField, err error) { +func (e *executor) executeMinRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ PairField, err0 error) { var filter *Row if len(c.Children) == 1 { @@ -1724,8 +1743,12 @@ func (e *executor) executeMinRowShard(ctx context.Context, qcx *Qcx, index strin } idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: fragment, Shard: fragment.shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: fragment, Shard: fragment.shard}) + if err != nil { + return PairField{}, err + } + + defer finisher(&err0) minRowID, count, err := fragment.minRow(tx, filter) if err != nil { @@ -1742,7 +1765,7 @@ func (e *executor) executeMinRowShard(ctx context.Context, qcx *Qcx, index strin } // executeMaxRowShard returns the maximum row ID for a shard. -func (e *executor) executeMaxRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ PairField, err error) { +func (e *executor) executeMaxRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ PairField, err0 error) { var filter *Row if len(c.Children) == 1 { @@ -1765,8 +1788,11 @@ func (e *executor) executeMaxRowShard(ctx context.Context, qcx *Qcx, index strin } idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return PairField{}, ErrQcxDone + } + defer finisher(&err0) maxRowID, count, err := fragment.maxRow(tx, filter) if err != nil { @@ -1935,7 +1961,10 @@ func (e *executor) executeTopNShard(ctx context.Context, qcx *Qcx, index string, } idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: !writable, Fragment: f, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Fragment: f, Index: idx, Shard: shard}) + if err != nil { + return nil, err + } defer finisher(&err0) pairs, err := f.top(tx, topOptions{ @@ -2764,7 +2793,10 @@ func (e *executor) executeRowsShard(ctx context.Context, qcx *Qcx, index string, filters = append(filters, filterLike(like, f.TranslateStore(), likeErr)) } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return nil, err + } defer finisher(&err0) for _, view := range views { @@ -3032,7 +3064,10 @@ func (e *executor) executeExtractShard(ctx context.Context, qcx *Qcx, index stri return ExtractedIDMatrix{}, newNotFoundError(ErrIndexNotFound, index) } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return ExtractedIDMatrix{}, err + } defer finisher(&err0) // Decompress columns bitmap. @@ -3260,7 +3295,10 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, return NewRow(), nil } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Fragment: frag, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Fragment: frag, Index: idx, Shard: shard}) + if err != nil { + return nil, err + } defer finisher(&err0) return frag.row(tx, rowID) } @@ -3286,7 +3324,10 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, if f == nil { continue } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Fragment: f, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Fragment: f, Index: idx, Shard: shard}) + if err != nil { + return nil, err + } defer finisher(&err0) row, err := f.row(tx, rowID) @@ -3334,7 +3375,10 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index return nil, newNotFoundError(ErrFieldNotFound, fieldName) } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: f.idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: f.idx, Shard: shard}) + if err != nil { + return nil, err + } defer finisher(&err0) // EQ null _exists - frag.NotNull() @@ -3554,7 +3598,7 @@ func (e *executor) executePrecomputedCallShard(ctx context.Context, qcx *Qcx, in } // executeNotShard executes a Not() call for a local shard. -func (e *executor) executeNotShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err error) { +func (e *executor) executeNotShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err0 error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeNotShard") defer span.Finish() @@ -3572,8 +3616,12 @@ func (e *executor) executeNotShard(ctx context.Context, qcx *Qcx, index string, return nil, errors.Errorf("index does not support existence tracking: %s", index) } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return nil, err + } + + defer finisher(&err0) var existenceRow *Row existenceFrag := e.Holder.fragment(index, existenceFieldName, viewStandard, shard) @@ -3687,7 +3735,7 @@ func (e *executor) executeUnionRows(ctx context.Context, qcx *Qcx, index string, } // executeAllCallShard executes an All() call for a local shard. -func (e *executor) executeAllCallShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (res *Row, err error) { +func (e *executor) executeAllCallShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (res *Row, err0 error) { span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeAllCallShard") defer span.Finish() @@ -3709,8 +3757,12 @@ func (e *executor) executeAllCallShard(ctx context.Context, qcx *Qcx, index stri if existenceFrag == nil { existenceRow = NewRow() } else { - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: existenceFrag, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: existenceFrag, Shard: shard}) + if err != nil { + return nil, err + } + + defer finisher(&err0) if existenceRow, err = existenceFrag.row(tx, 0); err != nil { return nil, err @@ -3821,7 +3873,7 @@ func (e *executor) executeClearBit(ctx context.Context, qcx *Qcx, index string, } // executeClearBitField executes a Clear() call for a field. -func (e *executor) executeClearBitField(ctx context.Context, qcx *Qcx, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *execOptions) (_ bool, err error) { +func (e *executor) executeClearBitField(ctx context.Context, qcx *Qcx, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *execOptions) (_ bool, err0 error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeClearBitField") defer span.Finish() @@ -3833,8 +3885,12 @@ func (e *executor) executeClearBitField(ctx context.Context, qcx *Qcx, index str if node.ID == e.Node.ID { idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } + + defer finisher(&err0) val, err := f.ClearBit(tx, rowID, colID) if err != nil { @@ -3911,12 +3967,11 @@ func (e *executor) executeClearRow(ctx context.Context, qcx *Qcx, index string, } // executeClearRowShard executes a ClearRow() call for a single shard. -func (e *executor) executeClearRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ bool, err error) { +func (e *executor) executeClearRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ bool, err0 error) { span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeClearRowShard") defer span.Finish() - var fieldName string - fieldName, err = c.FieldArg() + fieldName, err := c.FieldArg() if err != nil { return false, errors.New("ClearRow() argument required: field") } @@ -3937,8 +3992,11 @@ func (e *executor) executeClearRowShard(ctx context.Context, qcx *Qcx, index str } idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } + defer finisher(&err0) // Remove the row from all views. changed := false @@ -4010,9 +4068,9 @@ func (e *executor) executeSetRow(ctx context.Context, qcx *Qcx, indexName string } // executeSetRowShard executes a SetRow() call for a single shard. -func (e *executor) executeSetRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ bool, err error) { - var fieldName string - fieldName, err = c.FieldArg() +func (e *executor) executeSetRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ bool, err0 error) { + + fieldName, err := c.FieldArg() if err != nil { return false, errors.New("Store() argument required: field") } @@ -4060,8 +4118,12 @@ func (e *executor) executeSetRowShard(ctx context.Context, qcx *Qcx, index strin } idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } + + defer finisher(&err0) set, err := fragment.setRow(tx, src, rowID) if err != nil { @@ -4107,7 +4169,11 @@ func (e *executor) executeSet(ctx context.Context, qcx *Qcx, index string, c *pq // Set column on existence field. if ef := idx.existenceField(); ef != nil { // we create tx here, rather than just above, to avoid creating an extra empty shard. - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } + defer finisher(&err0) if _, err := ef.SetBit(tx, 0, colID, nil); err != nil { @@ -4176,7 +4242,10 @@ func (e *executor) executeSetBitField(ctx context.Context, qcx *Qcx, index strin if node.ID == e.Node.ID { idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } defer finisher(&err0) val, err := f.SetBit(tx, rowID, colID, timestamp) @@ -4204,7 +4273,7 @@ func (e *executor) executeSetBitField(ctx context.Context, qcx *Qcx, index strin } // executeSetValueField executes a Set() call for a specific int field. -func (e *executor) executeSetValueField(ctx context.Context, qcx *Qcx, index string, c *pql.Call, f *Field, colID uint64, value int64, opt *execOptions) (_ bool, err error) { +func (e *executor) executeSetValueField(ctx context.Context, qcx *Qcx, index string, c *pql.Call, f *Field, colID uint64, value int64, opt *execOptions) (_ bool, err0 error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSetValueField") defer span.Finish() @@ -4216,8 +4285,12 @@ func (e *executor) executeSetValueField(ctx context.Context, qcx *Qcx, index str if node.ID == e.Node.ID { idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } + + defer finisher(&err0) val, err := f.SetValue(tx, colID, value) if err != nil { @@ -4244,7 +4317,7 @@ func (e *executor) executeSetValueField(ctx context.Context, qcx *Qcx, index str } // executeClearValueField removes value for colID if present -func (e *executor) executeClearValueField(ctx context.Context, qcx *Qcx, index string, c *pql.Call, f *Field, colID uint64, opt *execOptions) (_ bool, err error) { +func (e *executor) executeClearValueField(ctx context.Context, qcx *Qcx, index string, c *pql.Call, f *Field, colID uint64, opt *execOptions) (_ bool, err0 error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeClearValueField") defer span.Finish() @@ -4256,8 +4329,11 @@ func (e *executor) executeClearValueField(ctx context.Context, qcx *Qcx, index s if node.ID == e.Node.ID { idx := e.Holder.Index(index) - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard}) + if err != nil { + return false, err + } + defer finisher(&err0) val, err := f.ClearValue(tx, colID) if err != nil { @@ -6332,7 +6408,7 @@ type groupByIterator struct { } // newGroupByIterator initializes a new groupByIterator. -func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children []*pql.Call, aggregate *pql.Call, filter *Row, index string, shard uint64, holder *Holder) (_ *groupByIterator, err error) { +func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children []*pql.Call, aggregate *pql.Call, filter *Row, index string, shard uint64, holder *Holder) (_ *groupByIterator, err0 error) { gbi := &groupByIterator{ executor: executor, @@ -6389,8 +6465,11 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children filters = append(filters, filterWithRows(rowIDs[i])) } - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) - defer finisher(&err) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return nil, err + } + defer finisher(&err0) gbi.rowIters[i], err = frag.rowIterator(tx, i != 0, filters...) if err != nil { diff --git a/field.go b/field.go index 7e747d672..970f4b6f6 100644 --- a/field.go +++ b/field.go @@ -1686,7 +1686,10 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []*time. return errors.Wrap(err, "creating fragment") } - tx, finisher := qcx.GetTx(Txo{Write: true, Index: frag.idx, Fragment: frag, Shard: frag.shard}) + 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) if err1 != nil { @@ -1798,12 +1801,15 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option } // now we know which shard we discovered. - tx, finisher := qcx.GetTx(Txo{Write: writable, Index: f.idx, Shard: frag.shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: f.idx, Shard: frag.shard}) + if err != nil { + return err + } // by deferring, even though we are in loop, we get en-mass commit at once if they all succeed, // or en-mass rollback if any fail. defer finisher(&err0) - if err := frag.importValue(tx, data.ColumnIDs, baseValues, requiredDepth, options.Clear); err != nil { + if err = frag.importValue(tx, data.ColumnIDs, baseValues, requiredDepth, options.Clear); err != nil { return err } } diff --git a/txfactory.go b/txfactory.go index 4f42ada64..e69f77cf2 100644 --- a/txfactory.go +++ b/txfactory.go @@ -211,6 +211,8 @@ func (f *TxFactory) NewQcx() (qcx *Qcx) { var NoopFinisher = func(perr *error) {} +var ErrQcxDone = fmt.Errorf("Qcx already Aborted or Finished, so must call reset before re-use") + // GetTx is used like this: // // someFunc(ctx context.Context, shard uint64) (_ interface{}, err0 error) { @@ -244,13 +246,12 @@ var NoopFinisher = func(perr *error) {} // be clearer (and much safer) to rename the enclosing functions 'err' to 'err0', // to make it clear we are referring to the first and final error. // -func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error)) { +func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) { qcx.mu.Lock() defer qcx.mu.Unlock() if qcx.done { - panic(fmt.Sprintf("cannot call GetTx on a Qcx that is already done. "+ - "Must call Reset() or txf.NewQcx(); stack='%v'", stack())) + return nil, nil, ErrQcxDone } // Use direct option if set on QCX. @@ -263,7 +264,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error)) { // roaring Tx are No-ops anyway, so just give it a new Tx // everytime. if qcx.isRoaring { - return qcx.Txf.NewTx(o), NoopFinisher + return qcx.Txf.NewTx(o), NoopFinisher, nil } // qcx.write reflects the top executor determination @@ -303,7 +304,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error)) { if o.Index.name != ro.Index.name { panic(fmt.Sprintf("index mismatch: o.Index = %v while qcx.RequiredTxo.Index = %v", o.Index.name, ro.Index.name)) } - return *qcx.RequiredForAtomicWriteTx, NoopFinisher + return *qcx.RequiredForAtomicWriteTx, NoopFinisher, nil } if !o.Write && qcx.Grp != nil { diff --git a/txfactory_internal_test.go b/txfactory_internal_test.go index 293e6fe21..6fa9c3b9e 100644 --- a/txfactory_internal_test.go +++ b/txfactory_internal_test.go @@ -53,7 +53,8 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) { // add to the group txn on the txf. qcx := idx.holder.txf.NewQcx() - tx, finisher := qcx.GetTx(Txo{Write: true, Index: idx, Shard: f.shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: true, Index: idx, Shard: f.shard}) + panicOn(err) // Set bits on the fragment. if _, err := f.setBit(tx, 120, 1); err != nil { @@ -76,7 +77,8 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) { panicOn(qcx.Finish()) qcx.Reset() - tx, finread := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard}) + tx, finread, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard}) + panicOn(err) if n := f.mustRow(tx, 120).Count(); n != 2 { panic(fmt.Sprintf("unexpected count (reopen): %d", n)) } else if n := f.mustRow(tx, 121).Count(); n != 1 { diff --git a/view.go b/view.go index ef39de39a..91e144cc4 100644 --- a/view.go +++ b/view.go @@ -537,7 +537,10 @@ func (v *view) rangeOp(qcx *Qcx, op pql.Token, bitDepth uint, predicate int64) ( r := NewRow() for _, frag := range v.allFragments() { - tx, finisher := qcx.GetTx(Txo{Write: !writable, Index: v.idx, Shard: frag.shard}) + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: v.idx, Shard: frag.shard}) + if err != nil { + return nil, err + } defer finisher(&err0) other, err := frag.rangeOp(tx, op, bitDepth, predicate)