mvcc_history_test.go

// Copyright 2019 The Cockroach Authors.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.

package engine

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"strconv"
	"strings"
	"testing"

	"github.com/cockroachdb/cockroach/pkg/roachpb"
	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
	"github.com/cockroachdb/cockroach/pkg/util/hlc"
	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
	"github.com/cockroachdb/cockroach/pkg/util/uint128"
	"github.com/cockroachdb/cockroach/pkg/util/uuid"
	"github.com/cockroachdb/datadriven"
	"github.com/cockroachdb/errors"
)

// TestMVCCHistories verifies that sequences of MVCC reads and writes
// perform properly.
func TestMVCCHistories(t *testing.T) {
	defer leaktest.AfterTest(t)()

	ctx := context.Background()
	for _, engineImpl := range mvccEngineImpls {
		t.Run(engineImpl.name, func(t *testing.T) {

			// Everything reads/writes under the same prefix.
			key := roachpb.Key("")
			span := roachpb.Span{Key: key, EndKey: key.PrefixEnd()}

			datadriven.Walk(t, "testdata/mvcc_histories", func(t *testing.T, path string) {
				// We start from a clean slate in every test file.
				engine := engineImpl.create()
				defer engine.Close()

				reportDataEntries := func(buf *bytes.Buffer) error {
					hasData := false
					err := engine.Iterate(
						span.Key,
						span.EndKey,
						func(r MVCCKeyValue) (bool, error) {
							hasData = true
							if r.Key.Timestamp.IsEmpty() {
								// Meta is at timestamp zero.
								meta := enginepb.MVCCMetadata{}
								if err := protoutil.Unmarshal(r.Value, &meta); err != nil {
									fmt.Fprintf(buf, "meta: %v -> error decoding proto from %v: %v\n", r.Key, r.Value, err)
								} else {
									fmt.Fprintf(buf, "meta: %v -> %+v\n", r.Key, &meta)
								}
							} else {
								fmt.Fprintf(buf, "data: %v -> %s\n", r.Key, roachpb.Value{RawBytes: r.Value}.PrettyPrint())
							}
							return false, nil
						})
					if !hasData {
						buf.WriteString("<no data>\n")
					}
					return err
				}

				e := newEvalCtx(ctx, engine)

				datadriven.RunTest(t, path, func(t *testing.T, d *datadriven.TestData) string {
					// We'll be overriding cmd/cmdargs below, because the
					// datadriven reader does not know about sub-commands.
					defer func(pos, cmd string, cmdArgs []datadriven.CmdArg) {
						d.Pos = pos
						d.Cmd = cmd
						d.CmdArgs = cmdArgs
					}(d.Pos, d.Cmd, d.CmdArgs)
					// The various evalCtx helpers want access to the current test
					// and testdata structs.
					e.t = t
					e.td = d

					switch d.Cmd {
					case "skip":
						if len(d.CmdArgs) == 0 || d.CmdArgs[0].Key == engineImpl.name {
							e.t.Skip("skipped")
						}
						return d.Expected
					case "run":
						// Syntax: run [trace] [error]
						// (other words - in particular "ok" - are accepted but ignored)
						//
						// "run" executes a script of zero or more operations from
						// the commands library defined below.
						// It stops upon the first error encountered, if any.
						//
						// Options:
						// "trace" means detail each operation in the output.
						// "error" means expect an error to occur. The specific error type/
						// message to expect is spelled out in the expected output.
						//
						trace := false
						if e.hasArg("trace") {
							trace = true
						}
						expectError := false
						if e.hasArg("error") {
							expectError = true
						}

						// buf will accumulate the actual output, which the
						// datadriven driver will use to compare to the expected
						// output.
						var buf bytes.Buffer
						e.results.buf = &buf

						// foundErr remembers which error was last encountered while
						// executing the script under "run".
						var foundErr error

						// pos is the original <file>:<lineno> prefix computed by
						// datadriven. It points to the top "run" command itself.
						// We editing d.Pos in-place below by extending `pos` upon
						// each new line of the script.
						pos := d.Pos

						// dataChange indicates whether some command in the script
						// has modified the stored data. When this becomes true, the
						// current content of storage is printed in the results
						// buffer at the end.
						dataChange := false
						// txnChange indicates whether some command has modified
						// a transaction object. When set, the last modified txn
						// object is reported in the result buffer at the end.
						txnChange := false

						reportResults := func(printTxn, printData bool) {
							if printTxn && e.results.txn != nil {
								fmt.Fprintf(&buf, "txn: %v\n", e.results.txn)
							}
							if printData {
								err := reportDataEntries(&buf)
								if err != nil {
									if foundErr == nil {
										// Handle the error below.
										foundErr = err
									} else {
										fmt.Fprintf(&buf, "error reading data: (%T:) %v\n", err, err)
									}
								}
							}
						}

						// sharedCmdArgs is updated by "with" pseudo-commands,
						// to pre-populate common arguments for the following
						// indented commands.
						var sharedCmdArgs []datadriven.CmdArg

						// The lines of the script under "run".
						lines := strings.Split(d.Input, "\n")
						for i, line := range lines {
							if short := strings.TrimSpace(line); short == "" || strings.HasPrefix(short, "#") {
								// Comment or empty line. Do nothing.
								continue
							}

							// Compute a line prefix, to clarify error message. We
							// prefix a newline character because some text editor do
							// not know how to jump to the location of an error if
							// there are multiple file:line prefixes on the same line.
							d.Pos = fmt.Sprintf("\n%s: (+%d)", pos, i+1)

							// Trace the execution in testing.T, to clarify where we
							// are in case an error occurs.
							e.t.Logf("%s: %s", d.Pos, line)

							// Decompose the current script line.
							var err error
							d.Cmd, d.CmdArgs, err = datadriven.ParseLine(line)
							if err != nil {
								e.t.Fatalf("%s: %v", d.Pos, err)
							}

							// Expand "with" commands:
							//   with t=A
							//       begin_txn
							//       resolve_intent k=a
							// is equivalent to:
							//   begin_txn      t=A
							//   resolve_intent k=a t=A
							isIndented := strings.TrimLeft(line, " \t") != line
							if d.Cmd == "with" {
								if !isIndented {
									// Reset shared args.
									sharedCmdArgs = d.CmdArgs
								} else {
									// Prefix shared args. We use prefix so that the
									// innermost "with" can override/shadow the outermost
									// "with".
									sharedCmdArgs = append(d.CmdArgs, sharedCmdArgs...)
								}
								continue
							} else if isIndented {
								// line is indented. Inherit arguments.
								if len(sharedCmdArgs) == 0 {
									// sanity check.
									e.Fatalf("indented command without prior 'with': %s", line)
								}
								// We prepend the args that are provided on the command
								// itself so it's possible to override those provided
								// via "with".
								d.CmdArgs = append(d.CmdArgs, sharedCmdArgs...)
							} else {
								// line is not indented. Clear shared arguments.
								sharedCmdArgs = nil
							}

							cmd := e.getCmd()
							txnChange = txnChange || cmd.txn
							dataChange = dataChange || cmd.data

							if trace {
								// If tracing is also requested by the datadriven input,
								// we'll trace the statement in the actual results too.
								fmt.Fprintf(&buf, ">> %s", d.Cmd)
								for i := range d.CmdArgs {
									fmt.Fprintf(&buf, " %s", &d.CmdArgs[i])
								}
								buf.WriteByte('\n')
							}

							// Run the command.
							foundErr = cmd.fn(e)

							if trace {
								// If tracing is enabled, we report the intermediate results
								// after each individual step in the script.
								// This may modify foundErr too.
								reportResults(cmd.txn, cmd.data)
							}

							if foundErr != nil {
								// An error occurred. Stop the script prematurely.
								break
							}
						}
						// End of script.

						if !trace {
							// If we were not tracing, no results were printed yet. Do it now.
							if txnChange || dataChange {
								buf.WriteString(">> at end:\n")
							}
							reportResults(txnChange, dataChange)
						}

						signalError := e.t.Errorf
						if txnChange || dataChange {
							// We can't recover from an error and continue
							// to proceed further tests, because the state
							// may have changed from what the test may be expecting.
							signalError = e.t.Fatalf
						}

						// Check for errors.
						if foundErr == nil && expectError {
							signalError("%s: expected error, got success", d.Pos)
							return d.Expected
						} else if foundErr != nil {
							if expectError {
								fmt.Fprintf(&buf, "error: (%T:) %v\n", foundErr, foundErr)
							} else /* !expectError */ {
								signalError("%s: expected success, found: (%T:) %v", d.Pos, foundErr, foundErr)
								return d.Expected
							}
						}

						// We're done. Report the actual results and errors to the
						// datadriven executor.
						return buf.String()

					default:
						e.t.Errorf("%s: unknown command: %s", d.Pos, d.Cmd)
						return d.Expected
					}
				})
			})
		})
	}
}

// getCmd retrieves the cmd entry for the current script line.
func (e *evalCtx) getCmd() cmd {
	e.t.Helper()
	c, ok := commands[e.td.Cmd]
	if !ok {
		e.Fatalf("unknown command: %s", e.td.Cmd)
	}
	return c
}

// cmd represents one supported script command.
type cmd struct {
	txn, data bool
	fn        func(e *evalCtx) error
}

// commands is the list of all supported script commands.
var commands = map[string]cmd{
	"begin_txn":      {true, false, cmdBegin},
	"remove_txn":     {true, false, cmdRemove},
	"advance_txn":    {true, false, cmdAdvanceTxn},
	"resolve_intent": {false, true, cmdResolve},
	"clear_range":    {false, true, cmdClearRange},
	"update_txn":     {true, false, cmdUpdateTxn},
	"restart_txn":    {true, false, cmdRestartTxn},
	"step_txn":       {true, false, cmdStepTxn},
	"txn_status":     {true, false, cmdSetTxnStatus},
	"cput":           {false, true, cmdCPut},
	"put":            {false, true, cmdPut},
	"merge":          {false, true, cmdMerge},
	"del":            {false, true, cmdDelete},
	"get":            {false, false, cmdGet},
	"scan":           {false, false, cmdScan},
	"increment":      {false, true, cmdIncrement},
	"check_intent":   {false, false, cmdCheckIntent},
}

func cmdClearRange(e *evalCtx) error {
	key, endKey := e.getKeyRange()
	return e.engine.ClearRange(
		MVCCKey{Key: key},
		MVCCKey{Key: endKey},
	)
}

func cmdBegin(e *evalCtx) error {
	var txnName string
	e.scanArg("t", &txnName)
	ts := e.getTs(nil)
	key := roachpb.KeyMin
	if e.hasArg("k") {
		key = e.getKey()
	}
	txn, err := e.newTxn(txnName, ts, key)
	e.results.txn = txn
	return err
}

func cmdResolve(e *evalCtx) error {
	txn := e.getTxn(mandatory)
	key := e.getKey()
	status := e.getTxnStatus()
	return e.resolveIntent(e.engine, key, txn, status)
}

func (e *evalCtx) resolveIntent(
	engine ReadWriter,
	key roachpb.Key,
	txn *roachpb.Transaction,
	resolveStatus roachpb.TransactionStatus,
) error {
	return MVCCResolveWriteIntent(e.ctx, engine, nil, roachpb.Intent{
		Span:   roachpb.Span{Key: key},
		Status: resolveStatus,
		Txn:    txn.TxnMeta,
	})
}

func cmdSetTxnStatus(e *evalCtx) error {
	txn := e.getTxn(mandatory)
	status := e.getTxnStatus()
	txn.Status = status
	e.results.txn = txn
	return nil
}

func cmdRemove(e *evalCtx) error {
	txn := e.getTxn(mandatory)
	delete(e.txns, txn.Name)
	e.results.txn = nil
	return nil
}

func cmdUpdateTxn(e *evalCtx) error {
	txn := e.getTxn(mandatory)
	var txnName2 string
	e.scanArg("t2", &txnName2)
	txn2, err := e.lookupTxn(txnName2)
	if err != nil {
		e.Fatalf("%v", err)
	}
	txn.Update(txn2)
	e.results.txn = txn
	return nil
}

func cmdGet(e *evalCtx) error {
	txn := e.getTxn(optional)
	key := e.getKey()
	ts := e.getTs(txn)
	opts := MVCCGetOptions{Txn: txn}
	if e.hasArg("inconsistent") {
		opts.Inconsistent = true
		opts.Txn = nil
	}
	if e.hasArg("tombstones") {
		opts.Tombstones = true
	}
	val, intent, err := MVCCGet(e.ctx, e.engine, key, ts, opts)
	if err != nil {
		return err
	}
	if intent != nil {
		fmt.Fprintf(e.results.buf, "get: %v -> intent {%s} %s\n", key, intent.Txn, intent.Status)
	}
	if val != nil {
		fmt.Fprintf(e.results.buf, "get: %v -> %v @%v\n", key, val.PrettyPrint(), val.Timestamp)
	} else {
		fmt.Fprintf(e.results.buf, "get: %v -> <no data>\n", key)
	}
	return nil
}

func cmdCheckIntent(e *evalCtx) error {
	key := e.getKey()
	wantIntent := true
	if e.hasArg("none") {
		wantIntent = false
	}
	metaKey := mvccKey(key)
	var meta enginepb.MVCCMetadata
	ok, _, _, err := e.engine.GetProto(metaKey, &meta)
	if err != nil {
		return err
	}
	if !ok && wantIntent {
		return errors.Newf("meta: %v -> expected intent, found none", key)
	}
	if ok {
		fmt.Fprintf(e.results.buf, "meta: %v -> %s\n", key, &meta)
		if !wantIntent {
			return errors.Newf("meta: %v -> expected no intent, found one", key)
		}
	}
	return nil
}

func cmdScan(e *evalCtx) error {
	txn := e.getTxn(optional)
	key, endKey := e.getKeyRange()
	ts := e.getTs(txn)
	opts := MVCCScanOptions{Txn: txn}
	if e.hasArg("inconsistent") {
		opts.Inconsistent = true
		opts.Txn = nil
	}
	if e.hasArg("tombstones") {
		opts.Tombstones = true
	}
	if e.hasArg("reverse") {
		opts.Reverse = true
	}
	max := int64(-1)
	if e.hasArg("max") {
		var imax int
		e.scanArg("max", &imax)
		max = int64(imax)
	}
	vals, _, intents, err := MVCCScan(e.ctx, e.engine, key, endKey, max, ts, opts)
	if err != nil {
		return err
	}
	for _, intent := range intents {
		fmt.Fprintf(e.results.buf, "scan: %v -> intent {%s} %s\n", key, intent.Txn, intent.Status)
	}
	for _, val := range vals {
		fmt.Fprintf(e.results.buf, "scan: %v -> %v @%v\n", val.Key, val.Value.PrettyPrint(), val.Value.Timestamp)
	}
	if len(vals) == 0 {
		fmt.Fprintf(e.results.buf, "scan: %v-%v -> <no data>\n", key, endKey)
	}
	return nil
}

func cmdPut(e *evalCtx) error {
	txn := e.getTxn(optional)
	ts := e.getTs(txn)

	key := e.getKey()
	val := e.getVal()

	resolve, resolveStatus := e.getResolve()

	return e.withWriter("put", func(engine ReadWriter) error {
		if err := MVCCPut(e.ctx, engine, nil, key, ts, val, txn); err != nil {
			return err
		}
		if resolve {
			return e.resolveIntent(engine, key, txn, resolveStatus)
		}
		return nil
	})
}

func cmdIncrement(e *evalCtx) error {
	txn := e.getTxn(optional)
	ts := e.getTs(txn)

	key := e.getKey()
	inc := int64(1)
	if e.hasArg("inc") {
		var incI int
		e.scanArg("inc", &incI)
		inc = int64(incI)
	}

	resolve, resolveStatus := e.getResolve()

	return e.withWriter("put", func(engine ReadWriter) error {
		curVal, err := MVCCIncrement(e.ctx, engine, nil, key, ts, txn, inc)
		if err != nil {
			return err
		}
		fmt.Fprintf(e.results.buf, "inc: current value = %d\n", curVal)
		if resolve {
			return e.resolveIntent(engine, key, txn, resolveStatus)
		}
		return nil
	})
}

func cmdCPut(e *evalCtx) error {
	txn := e.getTxn(optional)
	ts := e.getTs(txn)

	key := e.getKey()
	val := e.getVal()
	// Condition val is optional.
	var expVal *roachpb.Value
	if e.hasArg("cond") {
		rexpVal := e.getValInternal("cond")
		expVal = &rexpVal
	}
	behavior := CPutFailIfMissing
	if e.hasArg("allow_missing") {
		behavior = CPutAllowIfMissing
	}
	resolve, resolveStatus := e.getResolve()

	return e.withWriter("cput", func(engine ReadWriter) error {
		if err := MVCCConditionalPut(e.ctx, engine, nil, key, ts, val, expVal, behavior, txn); err != nil {
			return err
		}
		if resolve {
			return e.resolveIntent(engine, key, txn, resolveStatus)
		}
		return nil
	})
}

func cmdMerge(e *evalCtx) error {
	key := e.getKey()
	var value string
	e.scanArg("v", &value)
	var val roachpb.Value
	if e.hasArg("raw") {
		val.RawBytes = []byte(value)
	} else {
		val.SetString(value)
	}
	ts := e.getTs(nil)
	return e.withWriter("merge", func(engine ReadWriter) error {
		return MVCCMerge(e.ctx, engine, nil, key, ts, val)
	})
}

func cmdDelete(e *evalCtx) error {
	txn := e.getTxn(optional)
	key := e.getKey()
	ts := e.getTs(txn)
	resolve, resolveStatus := e.getResolve()
	return e.withWriter("del", func(engine ReadWriter) error {
		if err := MVCCDelete(e.ctx, engine, nil, key, ts, txn); err != nil {
			return err
		}
		if resolve {
			return e.resolveIntent(engine, key, txn, resolveStatus)
		}
		return nil
	})
}

func cmdRestartTxn(e *evalCtx) error {
	txn := e.getTxn(mandatory)
	ts := e.getTs(txn)
	up := roachpb.NormalUserPriority
	tp := enginepb.MinTxnPriority
	txn.Restart(up, tp, ts)
	e.results.txn = txn
	return nil
}

func cmdStepTxn(e *evalCtx) error {
	txn := e.getTxn(mandatory)
	n := 1
	if e.hasArg("seq") {
		e.scanArg("seq", &n)
		txn.Sequence = enginepb.TxnSeq(n)
	} else {
		if e.hasArg("n") {
			e.scanArg("n", &n)
		}
		txn.Sequence += enginepb.TxnSeq(n)
	}
	e.results.txn = txn
	return nil
}

func cmdAdvanceTxn(e *evalCtx) error {
	txn := e.getTxn(mandatory)
	ts := e.getTs(txn)
	if ts.Less(txn.ReadTimestamp) {
		e.Fatalf("cannot advance txn to earlier (%s) than its ReadTimestamp (%s)",
			ts, txn.ReadTimestamp)
	}
	txn.WriteTimestamp = ts
	e.results.txn = txn
	return nil
}

func toKey(s string) roachpb.Key {
	switch {
	case len(s) > 0 && s[0] == '+':
		return roachpb.Key(s[1:]).Next()
	case len(s) > 0 && s[0] == '=':
		return roachpb.Key(s[1:])
	case len(s) > 0 && s[0] == '-':
		return roachpb.Key(s[1:]).PrefixEnd()
	default:
		return roachpb.Key(s)
	}
}

// evalCtx stored the current state of the environment of a running
// script.
type evalCtx struct {
	results struct {
		buf io.Writer
		txn *roachpb.Transaction
	}
	ctx        context.Context
	engine     Engine
	t          *testing.T
	td         *datadriven.TestData
	txns       map[string]*roachpb.Transaction
	txnCounter uint128.Uint128
}

func newEvalCtx(ctx context.Context, engine Engine) *evalCtx {
	return &evalCtx{
		ctx:        ctx,
		engine:     engine,
		txns:       make(map[string]*roachpb.Transaction),
		txnCounter: uint128.FromInts(0, 1),
	}
}

func (e *evalCtx) getTxnStatus() roachpb.TransactionStatus {
	status := roachpb.COMMITTED
	if e.hasArg("status") {
		var sn string
		e.scanArg("status", &sn)
		s, ok := roachpb.TransactionStatus_value[sn]
		if !ok {
			e.Fatalf("invalid status: %s", sn)
		}
		status = roachpb.TransactionStatus(s)
	}
	return status
}

func (e *evalCtx) scanArg(key string, dests ...interface{}) {
	e.t.Helper()
	e.td.ScanArgs(e.t, key, dests...)
}

func (e *evalCtx) hasArg(key string) bool {
	for _, c := range e.td.CmdArgs {
		if c.Key == key {
			return true
		}
	}
	return false
}

func (e *evalCtx) Fatalf(format string, args ...interface{}) {
	e.t.Helper()
	e.td.Fatalf(e.t, format, args...)
}

func (e *evalCtx) getResolve() (bool, roachpb.TransactionStatus) {
	e.t.Helper()
	if !e.hasArg("resolve") {
		return false, roachpb.PENDING
	}
	return true, e.getTxnStatus()
}

func (e *evalCtx) getTs(txn *roachpb.Transaction) hlc.Timestamp {
	var ts hlc.Timestamp
	if txn != nil {
		ts = txn.ReadTimestamp
	}
	if !e.hasArg("ts") {
		return ts
	}
	var tsS string
	e.scanArg("ts", &tsS)
	parts := strings.Split(tsS, ",")

	// Find the wall time part.
	tsW, err := strconv.ParseInt(parts[0], 10, 64)
	if err != nil {
		e.Fatalf("%v", err)
	}
	ts.WallTime = tsW

	// Find the logical part, if there is one.
	var tsL int64
	if len(parts) > 1 {
		tsL, err = strconv.ParseInt(parts[1], 10, 32)
		if err != nil {
			e.Fatalf("%v", err)
		}
	}
	ts.Logical = int32(tsL)
	return ts
}

type optArg int

const (
	optional optArg = iota
	mandatory
)

func (e *evalCtx) getTxn(opt optArg) *roachpb.Transaction {
	e.t.Helper()
	if opt == optional && (e.hasArg("notxn") || !e.hasArg("t")) {
		return nil
	}
	var txnName string
	e.scanArg("t", &txnName)
	txn, err := e.lookupTxn(txnName)
	if err != nil {
		e.Fatalf("%v", err)
	}
	return txn
}

func (e *evalCtx) withWriter(cmd string, fn func(engine ReadWriter) error) error {
	var rw ReadWriter
	rw = e.engine
	var batch Batch
	if e.hasArg("batched") {
		batch = e.engine.NewBatch()
		defer batch.Close()
		rw = batch
	}
	origErr := fn(rw)
	if batch != nil {
		batchStatus := "non-empty"
		if batch.Empty() {
			batchStatus = "empty"
		}
		fmt.Fprintf(e.results.buf, "%s: batch after write is %s\n", cmd, batchStatus)
	}
	if origErr != nil {
		return origErr
	}
	if batch != nil {
		return batch.Commit(true)
	}
	return nil
}

func (e *evalCtx) getVal() roachpb.Value { return e.getValInternal("v") }
func (e *evalCtx) getValInternal(argName string) roachpb.Value {
	var value string
	e.scanArg(argName, &value)
	var val roachpb.Value
	if e.hasArg("raw") {
		val.RawBytes = []byte(value)
	} else {
		val.SetString(value)
	}
	return val
}

func (e *evalCtx) getKey() roachpb.Key {
	e.t.Helper()
	var keyS string
	e.scanArg("k", &keyS)
	return toKey(keyS)
}

func (e *evalCtx) getKeyRange() (sk, ek roachpb.Key) {
	e.t.Helper()
	var keyS string
	e.scanArg("k", &keyS)
	sk = toKey(keyS)
	ek = sk.Next()
	if e.hasArg("end") {
		var endKeyS string
		e.scanArg("end", &endKeyS)
		ek = toKey(endKeyS)
	}
	return sk, ek
}

func (e *evalCtx) newTxn(
	txnName string, ts hlc.Timestamp, key roachpb.Key,
) (*roachpb.Transaction, error) {
	if _, ok := e.txns[txnName]; ok {
		e.Fatalf("txn %s already open", txnName)
	}
	txn := &roachpb.Transaction{
		TxnMeta: enginepb.TxnMeta{
			ID:             uuid.FromUint128(e.txnCounter),
			Key:            []byte(key),
			WriteTimestamp: ts,
			Sequence:       0,
		},
		Name:                    txnName,
		DeprecatedOrigTimestamp: ts,
		ReadTimestamp:           ts,
		Status:                  roachpb.PENDING,
	}
	e.txnCounter = e.txnCounter.Add(1)
	e.txns[txnName] = txn
	return txn, nil
}

func (e *evalCtx) lookupTxn(txnName string) (*roachpb.Transaction, error) {
	txn, ok := e.txns[txnName]
	if !ok {
		e.Fatalf("txn %s not open", txnName)
	}
	return txn, nil
}