func (p *planner) getTableNames(dbDesc *DatabaseDescriptor) (parser.QualifiedNames, *roachpb.Error) {
	prefix := MakeNameMetadataKey(dbDesc.ID, "")
	sr, pErr := p.txn.Scan(prefix, prefix.PrefixEnd(), 0)
	if pErr != nil {
		return nil, pErr
	}

	var qualifiedNames parser.QualifiedNames
	for _, row := range sr {
		_, tableName, err := encoding.DecodeStringAscending(
			bytes.TrimPrefix(row.Key, prefix), nil)
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		qname := &parser.QualifiedName{
			Base:     parser.Name(dbDesc.Name),
			Indirect: parser.Indirection{parser.NameIndirection(tableName)},
		}
		if err := qname.NormalizeTableName(""); err != nil {
			return nil, roachpb.NewError(err)
		}
		qualifiedNames = append(qualifiedNames, qname)
	}
	return qualifiedNames, nil
}

// sendRPC sends one or more RPCs to replicas from the supplied roachpb.Replica
// slice. First, replicas which have gossiped addresses are corralled (and
// rearranged depending on proximity and whether the request needs to go to a
// leader) and then sent via Send, with requirement that one RPC to a server
// must succeed. Returns an RPC error if the request could not be sent. Note
// that the reply may contain a higher level error and must be checked in
// addition to the RPC error.
func (ds *DistSender) sendRPC(ctx context.Context, rangeID roachpb.RangeID, replicas ReplicaSlice,
	order orderingPolicy, ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
	if len(replicas) == 0 {
		return nil, roachpb.NewError(noNodeAddrsAvailError{})
	}

	// TODO(pmattis): This needs to be tested. If it isn't set we'll
	// still route the request appropriately by key, but won't receive
	// RangeNotFoundErrors.
	ba.RangeID = rangeID

	// Set RPC opts with stipulation that one of N RPCs must succeed.
	rpcOpts := SendOptions{
		Ordering:        order,
		SendNextTimeout: defaultSendNextTimeout,
		Timeout:         base.NetworkTimeout,
		Context:         ctx,
	}
	tracing.AnnotateTrace()
	defer tracing.AnnotateTrace()

	reply, err := ds.rpcSend(rpcOpts, replicas, ba, ds.rpcContext)
	if err != nil {
		return nil, roachpb.NewError(err)
	}
	return reply, nil
}

// TestEndWriteRestartReadOnlyTransaction verifies that if
// a transaction writes, then restarts and turns read-only,
// an explicit EndTransaction call is still sent if retry-
// able didn't, regardless of whether there is an error
// or not.
func TestEndWriteRestartReadOnlyTransaction(t *testing.T) {
	defer leaktest.AfterTest(t)
	for _, success := range []bool{true, false} {
		expCalls := []roachpb.Method{roachpb.BeginTransaction, roachpb.Put, roachpb.EndTransaction}
		var calls []roachpb.Method
		db := newDB(newTestSender(func(ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
			calls = append(calls, ba.Methods()...)
			return ba.CreateReply(), nil
		}, nil))
		ok := false
		if pErr := db.Txn(func(txn *Txn) *roachpb.Error {
			if !ok {
				if pErr := txn.Put("consider", "phlebas"); pErr != nil {
					t.Fatal(pErr)
				}
				ok = true
				return roachpb.NewError(&roachpb.TransactionRetryError{}) // immediate txn retry
			}
			if !success {
				return roachpb.NewError(errors.New("aborting on purpose"))
			}
			return nil
		}); pErr == nil != success {
			t.Errorf("expected error: %t, got error: %v", !success, pErr)
		}
		if !reflect.DeepEqual(expCalls, calls) {
			t.Fatalf("expected %v, got %v", expCalls, calls)
		}
	}
}

// Send forwards the call to the single store. This is a poor man's
// version of kv.TxnCoordSender, but it serves the purposes of
// supporting tests in this package. Transactions are not supported.
// Since kv/ depends on storage/, we can't get access to a
// TxnCoordSender from here.
// TODO(tschottdorf): {kv->storage}.LocalSender
func (db *testSender) Send(ctx context.Context, ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
	if et, ok := ba.GetArg(roachpb.EndTransaction); ok {
		return nil, roachpb.NewError(util.Errorf("%s method not supported", et.Method()))
	}
	// Lookup range and direct request.
	key, endKey := keys.Range(ba)
	rng := db.store.LookupReplica(key, endKey)
	if rng == nil {
		return nil, roachpb.NewError(roachpb.NewRangeKeyMismatchError(key, endKey, nil))
	}
	ba.RangeID = rng.Desc().RangeID
	replica := rng.GetReplica()
	if replica == nil {
		return nil, roachpb.NewError(util.Errorf("own replica missing in range"))
	}
	ba.Replica = *replica
	br, pErr := db.store.Send(ctx, ba)
	if br != nil && br.Error != nil {
		panic(roachpb.ErrorUnexpectedlySet(db.store, br))
	}
	if pErr != nil {
		return nil, pErr
	}
	return br, nil
}

// getCachedDatabaseDesc looks up the database descriptor given its name in the
// descriptor cache.
func (p *planner) getCachedDatabaseDesc(name string) (*DatabaseDescriptor, *roachpb.Error) {
	if name == systemDB.Name {
		return &systemDB, nil
	}

	nameKey := databaseKey{name}
	nameVal := p.systemConfig.GetValue(nameKey.Key())
	if nameVal == nil {
		return nil, roachpb.NewUErrorf("database %q does not exist in system cache", name)
	}

	id, err := nameVal.GetInt()
	if err != nil {
		return nil, roachpb.NewError(err)
	}

	descKey := MakeDescMetadataKey(ID(id))
	descVal := p.systemConfig.GetValue(descKey)
	if descVal == nil {
		return nil, roachpb.NewUErrorf("database %q has name entry, but no descriptor in system cache", name)
	}

	desc := &Descriptor{}
	if err := descVal.GetProto(desc); err != nil {
		return nil, roachpb.NewError(err)
	}

	database := desc.GetDatabase()
	if database == nil {
		return nil, roachpb.NewErrorf("%q is not a database", name)
	}
	return database, roachpb.NewError(database.Validate())
}

// getDescriptorFromTargetList examines a TargetList and fetches the
// appropriate descriptor.
// TODO(marc): support multiple targets.
func (p *planner) getDescriptorFromTargetList(targets parser.TargetList) (descriptorProto, *roachpb.Error) {
	if targets.Databases != nil {
		if len(targets.Databases) == 0 {
			return nil, roachpb.NewError(errNoDatabase)
		} else if len(targets.Databases) != 1 {
			return nil, roachpb.NewErrorf("TODO(marc): multiple targets not implemented")
		}
		descriptor, err := p.getDatabaseDesc(targets.Databases[0])
		if err != nil {
			return nil, err
		}
		return descriptor, nil
	}

	if len(targets.Tables) == 0 {
		return nil, roachpb.NewError(errNoTable)
	} else if len(targets.Tables) != 1 {
		return nil, roachpb.NewErrorf("TODO(marc): multiple targets not implemented")
	}
	descriptor, err := p.getTableDesc(targets.Tables[0])
	if err != nil {
		return nil, err
	}
	return descriptor, nil
}

// Prepare returns the result types of the given statement. Args may be a
// partially populated val args map. Prepare will populate the missing val
// args. The column result types are returned (or nil if there are no results).
func (e *Executor) Prepare(query string, session *Session, args parser.MapArgs) (
	[]ResultColumn, *roachpb.Error) {
	stmt, err := parser.ParseOne(query, parser.Syntax(session.Syntax))
	if err != nil {
		return nil, roachpb.NewError(err)
	}

	session.planner.resetForBatch(e)
	session.planner.evalCtx.Args = args
	session.planner.evalCtx.PrepareOnly = true

	// TODO(andrei): does the prepare phase really need a Txn?
	txn := client.NewTxn(*e.ctx.DB)
	txn.Proto.Isolation = session.DefaultIsolationLevel
	session.planner.setTxn(txn)
	defer session.planner.setTxn(nil)

	plan, pErr := session.planner.prepare(stmt)
	if pErr != nil {
		return nil, pErr
	}
	if plan == nil {
		return nil, nil
	}
	cols := plan.Columns()
	for _, c := range cols {
		if err := checkResultDatum(c.Typ); err != nil {
			return nil, roachpb.NewError(err)
		}
	}
	return cols, nil
}

func (p *planner) SetTimeZone(n *parser.SetTimeZone) (planNode, *roachpb.Error) {
	d, err := n.Value.Eval(p.evalCtx)
	if err != nil {
		return nil, roachpb.NewError(err)
	}
	var offset int64
	switch v := d.(type) {
	case parser.DString:
		location := string(v)
		if location == "DEFAULT" || location == "LOCAL" {
			location = "UTC"
		}
		if _, err := time.LoadLocation(location); err != nil {
			return nil, roachpb.NewError(err)
		}
		p.session.Timezone = &Session_Location{Location: location}

	case parser.DInterval:
		offset = int64(v.Duration / time.Second)

	case parser.DInt:
		offset = int64(v) * 60 * 60

	case parser.DFloat:
		offset = int64(float64(v) * 60.0 * 60.0)

	default:
		return nil, roachpb.NewUErrorf("bad time zone value: %v", n.Value)
	}
	if offset != 0 {
		p.session.Timezone = &Session_Offset{Offset: offset}
	}
	p.evalCtx.GetLocation = p.session.getLocation
	return &valuesNode{}, nil
}

// execStmtInAbortedTxn executes a statement in a txn that's in state
// Aborted or RestartWait.
// Everything but COMMIT/ROLLBACK/RESTART causes errors.
func (e *Executor) execStmtInAbortedTxn(
	stmt parser.Statement, txnState *txnState) (Result, *roachpb.Error) {
	if txnState.State != Aborted && txnState.State != RestartWait {
		panic("execStmtInAbortedTxn called outside of an aborted txn")
	}
	switch stmt.(type) {
	case *parser.CommitTransaction, *parser.RollbackTransaction:
		if txnState.State == RestartWait {
			if pErr := txnState.txn.Rollback(); pErr != nil {
				log.Errorf("failure rolling back transaction: %s", pErr)
			}
		}
		// Reset the state to allow new transactions to start.
		// Note: postgres replies to COMMIT of failed txn with "ROLLBACK" too.
		result := Result{PGTag: (*parser.RollbackTransaction)(nil).StatementTag()}
		txnState.resetStateAndTxn(NoTxn)
		return result, nil
	case *parser.RestartTransaction:
		if txnState.State == RestartWait {
			// Reset the state. Txn is Open again.
			txnState.State = Open
			// TODO(andrei/cdo): add a counter for user-directed retries.
			return Result{}, nil
		}
		pErr := roachpb.NewError(&roachpb.SqlTransactionAbortedError{
			CustomMsg: "RETRY INTENT has not been used or a non-retriable error was encountered."})
		return Result{PErr: pErr}, pErr
	default:
		pErr := roachpb.NewError(&roachpb.SqlTransactionAbortedError{})
		return Result{PErr: pErr}, pErr
	}
}

// getTableID retrieves the table ID for the specified table. It uses the
// descriptor cache to perform lookups, falling back to the KV store when
// necessary.
func (p *planner) getTableID(qname *parser.QualifiedName) (ID, *roachpb.Error) {
	if err := qname.NormalizeTableName(p.session.Database); err != nil {
		return 0, roachpb.NewError(err)
	}

	dbID, pErr := p.getDatabaseID(qname.Database())
	if pErr != nil {
		return 0, pErr
	}

	// Lookup the ID of the table in the cache. The use of the cache might cause
	// the usage of a recently renamed table, but that's a race that could occur
	// anyways.
	nameKey := tableKey{dbID, qname.Table()}
	key := nameKey.Key()
	if nameVal := p.systemConfig.GetValue(key); nameVal != nil {
		id, err := nameVal.GetInt()
		return ID(id), roachpb.NewError(err)
	}

	gr, pErr := p.txn.Get(key)
	if pErr != nil {
		return 0, pErr
	}
	if !gr.Exists() {
		return 0, roachpb.NewErrorf("table %q does not exist", nameKey.Name())
	}
	return ID(gr.ValueInt()), nil
}

func (n *scanNode) initWhere(where *parser.Where) *roachpb.Error {
	if where == nil {
		return nil
	}
	n.filter, n.pErr = n.resolveQNames(where.Expr)
	if n.pErr == nil {
		var whereType parser.Datum
		var err error
		whereType, err = n.filter.TypeCheck(n.planner.evalCtx.Args)
		n.pErr = roachpb.NewError(err)
		if n.pErr == nil {
			if !(whereType == parser.DummyBool || whereType == parser.DNull) {
				n.pErr = roachpb.NewUErrorf("argument of WHERE must be type %s, not type %s", parser.DummyBool.Type(), whereType.Type())
			}
		}
	}
	if n.pErr == nil {
		// Normalize the expression (this will also evaluate any branches that are
		// constant).
		var err error
		n.filter, err = n.planner.parser.NormalizeExpr(n.planner.evalCtx, n.filter)
		n.pErr = roachpb.NewError(err)
	}
	if n.pErr == nil {
		n.filter, n.pErr = n.planner.expandSubqueries(n.filter, 1)
	}
	return n.pErr
}

// RenameDatabase renames the database.
// Privileges: "root" user.
//   Notes: postgres requires superuser, db owner, or "CREATEDB".
//          mysql >= 5.1.23 does not allow database renames.
func (p *planner) RenameDatabase(n *parser.RenameDatabase) (planNode, *roachpb.Error) {
	if n.Name == "" || n.NewName == "" {
		return nil, roachpb.NewError(errEmptyDatabaseName)
	}

	if p.user != security.RootUser {
		return nil, roachpb.NewUErrorf("only %s is allowed to rename databases", security.RootUser)
	}

	dbDesc, pErr := p.getDatabaseDesc(string(n.Name))
	if pErr != nil {
		return nil, pErr
	}

	if n.Name == n.NewName {
		// Noop.
		return &valuesNode{}, nil
	}

	// Now update the nameMetadataKey and the descriptor.
	descKey := MakeDescMetadataKey(dbDesc.GetID())
	dbDesc.SetName(string(n.NewName))

	if err := dbDesc.Validate(); err != nil {
		return nil, roachpb.NewError(err)
	}

	newKey := databaseKey{string(n.NewName)}.Key()
	oldKey := databaseKey{string(n.Name)}.Key()
	descID := dbDesc.GetID()
	descDesc := wrapDescriptor(dbDesc)

	b := client.Batch{}
	b.CPut(newKey, descID, nil)
	b.Put(descKey, descDesc)
	b.Del(oldKey)

	if pErr := p.txn.Run(&b); pErr != nil {
		if _, ok := pErr.GoError().(*roachpb.ConditionFailedError); ok {
			return nil, roachpb.NewUErrorf("the new database name %q already exists", string(n.NewName))
		}
		return nil, pErr
	}

	p.testingVerifyMetadata = func(systemConfig config.SystemConfig) error {
		if err := expectDescriptorID(systemConfig, newKey, descID); err != nil {
			return err
		}
		if err := expectDescriptor(systemConfig, descKey, descDesc); err != nil {
			return err
		}
		return expectDeleted(systemConfig, oldKey)
	}

	return &valuesNode{}, nil
}

// initScan initializes but does not perform the key-value scan.
func (n *scanNode) initScan() bool {
	// Initialize our key/values.
	if len(n.spans) == 0 {
		// If no spans were specified retrieve all of the keys that start with our
		// index key prefix.
		start := roachpb.Key(MakeIndexKeyPrefix(n.desc.ID, n.index.ID))
		n.spans = append(n.spans, span{
			start: start,
			end:   start.PrefixEnd(),
		})
	}

	if n.valTypes == nil {
		// Prepare our index key vals slice.
		var err error
		n.valTypes, err = makeKeyVals(&n.desc, n.columnIDs)
		n.pErr = roachpb.NewError(err)
		if n.pErr != nil {
			return false
		}
		n.vals = make([]parser.Datum, len(n.valTypes))

		if n.isSecondaryIndex && n.index.Unique {
			// Unique secondary indexes have a value that is the primary index
			// key. Prepare implicitVals for use in decoding this value.
			// Primary indexes only contain ascendingly-encoded values. If this
			// ever changes, we'll probably have to figure out the directions here too.
			var err error
			n.implicitValTypes, err = makeKeyVals(&n.desc, n.index.ImplicitColumnIDs)
			n.pErr = roachpb.NewError(err)
			if n.pErr != nil {
				return false
			}
			n.implicitVals = make([]parser.Datum, len(n.implicitValTypes))
		}
	}

	// If we have a limit hint, we limit the first batch size. Subsequent batches use the normal
	// size, to avoid making things too slow (e.g. in case we have a very restrictive filter and
	// actually have to retrieve a lot of rows).
	firstBatchLimit := n.limitHint
	if firstBatchLimit != 0 {
		// For a secondary index, we have one key per row.
		if !n.isSecondaryIndex {
			// We have a sentinel key per row plus at most one key per non-PK column. Of course, we
			// may have other keys due to a schema change, but this is only a hint.
			firstBatchLimit *= int64(1 + len(n.visibleCols) - len(n.index.ColumnIDs))
		}
		// We need an extra key to make sure we form the last row.
		firstBatchLimit++
	}

	n.fetcher = makeKVFetcher(n.txn, n.spans, n.reverse, firstBatchLimit)
	n.scanInitialized = true
	return true
}

// RenameIndex renames the index.
// Privileges: CREATE on table.
//   notes: postgres requires CREATE on the table.
//          mysql requires ALTER, CREATE, INSERT on the table.
func (p *planner) RenameIndex(n *parser.RenameIndex) (planNode, *roachpb.Error) {
	newIdxName := string(n.NewName)
	if newIdxName == "" {
		return nil, roachpb.NewError(errEmptyIndexName)
	}

	if err := n.Name.NormalizeTableName(p.session.Database); err != nil {
		return nil, roachpb.NewError(err)
	}

	tableDesc, pErr := p.getTableDesc(n.Name)
	if pErr != nil {
		return nil, pErr
	}

	idxName := n.Name.Index()
	status, i, pErr := tableDesc.FindIndexByName(idxName)
	if pErr != nil {
		if n.IfExists {
			// Noop.
			return &valuesNode{}, nil
		}
		// Index does not exist, but we want it to: error out.
		return nil, pErr
	}

	if pErr := p.checkPrivilege(tableDesc, privilege.CREATE); pErr != nil {
		return nil, pErr
	}

	if equalName(idxName, newIdxName) {
		// Noop.
		return &valuesNode{}, nil
	}

	if _, _, err := tableDesc.FindIndexByName(newIdxName); err == nil {
		return nil, roachpb.NewUErrorf("index name %q already exists", n.NewName)
	}

	if status == DescriptorActive {
		tableDesc.Indexes[i].Name = newIdxName
	} else {
		tableDesc.Mutations[i].GetIndex().Name = newIdxName
	}

	tableDesc.UpVersion = true
	descKey := MakeDescMetadataKey(tableDesc.GetID())
	if err := tableDesc.Validate(); err != nil {
		return nil, roachpb.NewError(err)
	}
	if pErr := p.txn.Put(descKey, wrapDescriptor(tableDesc)); pErr != nil {
		return nil, pErr
	}
	p.notifySchemaChange(tableDesc.ID, invalidMutationID)
	return &valuesNode{}, nil
}

func (p *planner) query(sql string, args ...interface{}) (planNode, *roachpb.Error) {
	stmt, err := parser.ParseOneTraditional(sql)
	if err != nil {
		return nil, roachpb.NewError(err)
	}
	if err := parser.FillArgs(stmt, golangParameters(args)); err != nil {
		return nil, roachpb.NewError(err)
	}
	return p.makePlan(stmt, false)
}

// Initializes a scanNode with a tableName. Returns the table or index name that can be used for
// fully-qualified columns if an alias is not specified.
func (n *scanNode) initTable(p *planner, tableName *parser.QualifiedName) (string, *roachpb.Error) {
	if n.desc, n.pErr = p.getTableLease(tableName); n.pErr != nil {
		return "", n.pErr
	}

	if err := p.checkPrivilege(&n.desc, privilege.SELECT); err != nil {
		return "", roachpb.NewError(err)
	}

	alias := n.desc.Name

	indexName := tableName.Index()
	if indexName != "" && !equalName(n.desc.PrimaryIndex.Name, indexName) {
		for i := range n.desc.Indexes {
			if equalName(n.desc.Indexes[i].Name, indexName) {
				// Remove all but the matching index from the descriptor.
				n.desc.Indexes = n.desc.Indexes[i : i+1]
				n.index = &n.desc.Indexes[0]
				break
			}
		}
		if n.index == nil {
			n.pErr = roachpb.NewUErrorf("index \"%s\" not found", indexName)
			return "", n.pErr
		}
		// Use the index name instead of the table name for fully-qualified columns in the
		// expression.
		alias = n.index.Name
		// Strip out any columns from the table that are not present in the
		// index.
		visibleCols := make([]ColumnDescriptor, 0, len(n.index.ColumnIDs)+len(n.index.ImplicitColumnIDs))
		for _, colID := range n.index.ColumnIDs {
			col, err := n.desc.FindColumnByID(colID)
			n.pErr = roachpb.NewError(err)
			if n.pErr != nil {
				return "", n.pErr
			}
			visibleCols = append(visibleCols, *col)
		}
		for _, colID := range n.index.ImplicitColumnIDs {
			col, err := n.desc.FindColumnByID(colID)
			n.pErr = roachpb.NewError(err)
			if n.pErr != nil {
				return "", n.pErr
			}
			visibleCols = append(visibleCols, *col)
		}
		n.isSecondaryIndex = true
		n.initVisibleCols(visibleCols, len(n.index.ImplicitColumnIDs))
	} else {
		n.initDescDefaults()
	}

	return alias, nil
}

func (s *selectNode) addRender(target parser.SelectExpr) *roachpb.Error {
	// outputName will be empty if the target is not aliased.
	outputName := string(target.As)

	if isStar, cols, exprs, err := checkRenderStar(target, &s.table, s.qvals); err != nil {
		s.pErr = roachpb.NewError(err)
		return s.pErr
	} else if isStar {
		s.columns = append(s.columns, cols...)
		s.render = append(s.render, exprs...)
		return nil
	}

	// When generating an output column name it should exactly match the original
	// expression, so determine the output column name before we perform any
	// manipulations to the expression.
	outputName = getRenderColName(target)

	// Resolve qualified names. This has the side-effect of normalizing any
	// qualified name found.
	var resolved parser.Expr
	var err error
	if resolved, err = s.resolveQNames(target.Expr); err != nil {
		s.pErr = roachpb.NewError(err)
		return s.pErr
	}
	if resolved, s.pErr = s.planner.expandSubqueries(resolved, 1); s.pErr != nil {
		return s.pErr
	}
	var typ parser.Datum
	typ, err = resolved.TypeCheck(s.planner.evalCtx.Args)
	s.pErr = roachpb.NewError(err)
	if s.pErr != nil {
		return s.pErr
	}
	var normalized parser.Expr
	normalized, err = s.planner.parser.NormalizeExpr(s.planner.evalCtx, resolved)
	s.pErr = roachpb.NewError(err)
	if s.pErr != nil {
		return s.pErr
	}
	s.render = append(s.render, normalized)

	if target.As == "" {
		switch t := target.Expr.(type) {
		case *parser.QualifiedName:
			// If the expression is a qualified name, use the column name, not the
			// full qualification as the column name to return.
			outputName = t.Column()
		}
	}
	s.columns = append(s.columns, ResultColumn{Name: outputName, Typ: typ})
	return nil
}

// CreateTable creates a table.
// Privileges: CREATE on database.
//   Notes: postgres/mysql require CREATE on database.
func (p *planner) CreateTable(n *parser.CreateTable) (planNode, *roachpb.Error) {
	if err := n.Table.NormalizeTableName(p.session.Database); err != nil {
		return nil, roachpb.NewError(err)
	}

	dbDesc, pErr := p.getDatabaseDesc(n.Table.Database())
	if pErr != nil {
		return nil, pErr
	}

	if err := p.checkPrivilege(dbDesc, privilege.CREATE); err != nil {
		return nil, roachpb.NewError(err)
	}

	desc, pErr := makeTableDesc(n, dbDesc.ID)
	if pErr != nil {
		return nil, pErr
	}
	// Inherit permissions from the database descriptor.
	desc.Privileges = dbDesc.GetPrivileges()

	if len(desc.PrimaryIndex.ColumnNames) == 0 {
		// Ensure a Primary Key exists.
		s := "experimental_unique_int()"
		col := ColumnDescriptor{
			Name: "rowid",
			Type: ColumnType{
				Kind: ColumnType_INT,
			},
			DefaultExpr: &s,
			Hidden:      true,
			Nullable:    false,
		}
		desc.AddColumn(col)
		idx := IndexDescriptor{
			Unique:           true,
			ColumnNames:      []string{col.Name},
			ColumnDirections: []IndexDescriptor_Direction{IndexDescriptor_ASC},
		}
		if err := desc.AddIndex(idx, true); err != nil {
			return nil, roachpb.NewError(err)
		}
	}

	if pErr := desc.AllocateIDs(); pErr != nil {
		return nil, pErr
	}

	if pErr := p.createDescriptor(tableKey{dbDesc.ID, n.Table.Table()}, &desc, n.IfNotExists); pErr != nil {
		return nil, pErr
	}
	return &emptyNode{}, nil
}

// execStmtInOpenTxn executes one statement in the context
// of the planner's transaction (which is assumed to exist).
// It handles statements that affect the transaction state (BEGIN, COMMIT)
// and delegates everything else to `execStmt`.
// It binds placeholders.
//
// The current transaction might be committed/rolled back when this returns.
//
// Args:
// abortedMode: if set, we're in a transaction that has encountered errors, so we
//  must reject the statement unless it's a COMMIT/ROLLBACK.
// implicitTxn: set if the current transaction was implicitly
//  created by the system (i.e. the client sent the statement outside of
//  a transaction).
//  COMMIT/ROLLBACK statements are rejected if set. Also, the transaction
//  might be auto-committed in this function.
// firstInTxn: set for the first statement in a transaction. Used
//  so that nested BEGIN statements are caught.
// stmtTimestamp: Used as the statement_timestamp().
//
// Returns:
// - a Result
// - an error, if any. In case of error, the result returned also reflects this error.
func (e *Executor) execStmtInOpenTxn(
	stmt parser.Statement, planMaker *planner,
	implicitTxn bool,
	firstInTxn bool,
	stmtTimestamp parser.DTimestamp,
	txnState *txnState) (Result, *roachpb.Error) {
	if txnState.state() != openTransaction {
		panic("execStmtInOpenTxn called outside of an open txn")
	}
	if planMaker.txn == nil {
		panic("execStmtInOpenTxn called with the a txn not set on the planner")
	}

	planMaker.evalCtx.StmtTimestamp = stmtTimestamp

	// TODO(cdo): Figure out how to not double count on retries.
	e.updateStmtCounts(stmt)
	switch stmt.(type) {
	case *parser.BeginTransaction:
		if !firstInTxn {
			txnState.aborted = true
			pErr := roachpb.NewError(errTransactionInProgress)
			return Result{PErr: pErr}, pErr
		}
	case *parser.CommitTransaction, *parser.RollbackTransaction, *parser.SetTransaction:
		if implicitTxn {
			txnState.aborted = true
			pErr := roachpb.NewError(errNoTransactionInProgress)
			return Result{PErr: pErr}, pErr
		}
	}

	// Bind all the placeholder variables in the stmt to actual values.
	stmt, err := parser.FillArgs(stmt, &planMaker.params)
	if err != nil {
		txnState.aborted = true
		pErr := roachpb.NewError(err)
		return Result{PErr: pErr}, pErr
	}

	result, pErr := e.execStmt(stmt, planMaker, timeutil.Now(),
		implicitTxn /* autoCommit */)
	txnDone := planMaker.txn == nil
	if pErr != nil {
		result = Result{PErr: pErr}
		txnState.aborted = true
	}
	if txnDone {
		txnState.aborted = false
		txnState.txn = nil
	}
	return result, pErr
}

// Send implements the client.Sender interface. The store is looked up from the
// store map if specified by the request; otherwise, the command is being
// executed locally, and the replica is determined via lookup through each
// store's LookupRange method. The latter path is taken only by unit tests.
func (ls *Stores) Send(ctx context.Context, ba roachpb.BatchRequest) (*roachpb.BatchResponse, *roachpb.Error) {
	// If we aren't given a Replica, then a little bending over
	// backwards here. This case applies exclusively to unittests.
	if ba.RangeID == 0 || ba.Replica.StoreID == 0 {
		rs, err := keys.Range(ba)
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		rangeID, repl, err := ls.lookupReplica(rs.Key, rs.EndKey)
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		ba.RangeID = rangeID
		ba.Replica = *repl
	}

	ctx = log.Add(ctx,
		log.RangeID, ba.RangeID)

	store, err := ls.GetStore(ba.Replica.StoreID)
	if err != nil {
		return nil, roachpb.NewError(err)
	}

	if ba.Txn != nil {
		// For calls that read data within a txn, we keep track of timestamps
		// observed from the various participating nodes' HLC clocks. If we have
		// a timestamp on file for this Node which is smaller than MaxTimestamp,
		// we can lower MaxTimestamp accordingly. If MaxTimestamp drops below
		// OrigTimestamp, we effectively can't see uncertainty restarts any
		// more.
		// Note that it's not an issue if MaxTimestamp propagates back out to
		// the client via a returned Transaction update - when updating a Txn
		// from another, the larger MaxTimestamp wins.
		if maxTS, ok := ba.Txn.GetObservedTimestamp(ba.Replica.NodeID); ok && maxTS.Less(ba.Txn.MaxTimestamp) {
			// Copy-on-write to protect others we might be sharing the Txn with.
			shallowTxn := *ba.Txn
			// The uncertainty window is [OrigTimestamp, maxTS), so if that window
			// is empty, there won't be any uncertainty restarts.
			if !ba.Txn.OrigTimestamp.Less(maxTS) {
				log.Trace(ctx, "read has no clock uncertainty")
			}
			shallowTxn.MaxTimestamp.Backward(maxTS)
			ba.Txn = &shallowTxn
		}
	}
	br, pErr := store.Send(ctx, ba)
	if br != nil && br.Error != nil {
		panic(roachpb.ErrorUnexpectedlySet(store, br))
	}
	return br, pErr
}