// incCmd adds one to the value of c.key in the env and writes
// it to the db. If c.key isn't in the db, writes 1.
func incCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	r, pErr := txn.Inc(c.getKey(), 1)
	if pErr != nil {
		return pErr
	}
	c.env[c.key] = r.ValueInt()
	c.debug = fmt.Sprintf("[%d]", r.ValueInt())
	return nil
}

// incCmd adds one to the value of c.key in the env (as determined by
// a previous read or write, or else assumed to be zero) and writes it
// to the db.
func incCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	r := c.env[c.key] + 1
	if pErr := txn.Put(c.getKey(), r); pErr != nil {
		return pErr
	}
	c.env[c.key] = r
	c.debug = fmt.Sprintf("[%d]", r)
	return nil
}

// readCmd reads a value from the db and stores it in the env.
func readCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	r, pErr := txn.Get(c.getKey())
	if pErr != nil {
		return pErr
	}
	if r.Value != nil {
		c.env[c.key] = r.ValueInt()
		c.debug = fmt.Sprintf("[%d ts=%d]", r.ValueInt(), r.Timestamp())
	}
	return nil
}

// sumCmd sums the values of all keys != c.key read during the transaction and
// writes the result to the db.
func sumCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	sum := int64(0)
	for k, v := range c.env {
		if k != c.key {
			sum += v
		}
	}
	r, pErr := txn.Inc(c.getKey(), sum)
	c.debug = fmt.Sprintf("[%d ts=%d]", sum, r.Timestamp())
	return pErr
}

// GetRangeTree fetches the RangeTree proto and sets up the range tree context.
func getRangeTree(txn *client.Txn) (*treeContext, error) {
	tree := &proto.RangeTree{}
	if err := txn.GetProto(keys.RangeTreeRoot, tree); err != nil {
		return nil, err
	}
	return &treeContext{
		txn:   txn,
		tree:  tree,
		dirty: false,
		nodes: map[string]cachedNode{},
	}, nil
}

// GetRangeTree fetches the RangeTree proto and sets up the range tree context.
func getRangeTree(txn *client.Txn) (*treeContext, *roachpb.Error) {
	tree := new(RangeTree)
	if pErr := txn.GetProto(keys.RangeTreeRoot, tree); pErr != nil {
		return nil, pErr
	}
	return &treeContext{
		txn:   txn,
		tree:  tree,
		dirty: false,
		nodes: map[string]cachedNode{},
	}, nil
}

// incCmd adds one to the value of c.key in the env (as determined by
// a previous read or write, or else assumed to be zero) and writes it
// to the db.
func incCmd(c *cmd, txn *client.Txn, t *testing.T) error {
	val, ok := c.env[c.key]
	if !ok {
		panic(fmt.Sprintf("can't increment key %q; not yet read", c.key))
	}
	r := val + 1
	if err := txn.Put(c.getKey(), r); err != nil {
		return err
	}
	c.env[c.key] = r
	c.debug = fmt.Sprintf("[%d]", r)
	return nil
}

// readCmd reads a value from the db and stores it in the env.
func readCmd(c *cmd, txn *client.Txn, t *testing.T) error {
	r, err := txn.Get(c.getKey())
	if err != nil {
		return err
	}
	var value int64
	if r.Value != nil {
		value = r.ValueInt()
	}
	c.env[c.key] = value
	c.debug = fmt.Sprintf("[%d]", value)
	return nil
}

// writeCmd sums values from the env (and possibly numeric constants)
// and writes the value to the db. "c.endKey" here needs to be parsed
// in the context of this command, which is a "+"-separated list of
// keys from the env or numeric constants to sum.
func writeCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	sum := int64(0)
	for _, sp := range strings.Split(c.endKey, "+") {
		if constant, err := strconv.Atoi(sp); err != nil {
			sum += c.env[sp]
		} else {
			sum += int64(constant)
		}
	}
	pErr := txn.Put(c.getKey(), sum)
	c.debug = fmt.Sprintf("[%d]", sum)
	return pErr
}

// getTableDescFromID retrieves the table descriptor for the table
// ID passed in using an existing txn. Teturns an error if the
// descriptor doesn't exist or if it exists and is not a table.
func getTableDescFromID(txn *client.Txn, id sqlbase.ID) (*sqlbase.TableDescriptor, error) {
	desc := &sqlbase.Descriptor{}
	descKey := sqlbase.MakeDescMetadataKey(id)

	if err := txn.GetProto(descKey, desc); err != nil {
		return nil, err
	}
	table := desc.GetTable()
	if table == nil {
		return nil, errDescriptorNotFound
	}
	return table, nil
}

// truncateTable truncates the data of a table.
// It deletes a range of data for the table, which includes the PK and all
// indexes.
func truncateTable(tableDesc *sqlbase.TableDescriptor, txn *client.Txn) error {
	tablePrefix := keys.MakeTablePrefix(uint32(tableDesc.ID))

	// Delete rows and indexes starting with the table's prefix.
	tableStartKey := roachpb.Key(tablePrefix)
	tableEndKey := tableStartKey.PrefixEnd()
	if log.V(2) {
		log.Infof("DelRange %s - %s", tableStartKey, tableEndKey)
	}
	b := client.Batch{}
	b.DelRange(tableStartKey, tableEndKey, false)
	return txn.Run(&b)
}

// get the table descriptor for the ID passed in using the planner's txn.
func getTableDescFromID(txn *client.Txn, id ID) (*TableDescriptor, *roachpb.Error) {
	desc := &Descriptor{}
	descKey := MakeDescMetadataKey(id)

	if pErr := txn.GetProto(descKey, desc); pErr != nil {
		return nil, pErr
	}
	tableDesc := desc.GetTable()
	if tableDesc == nil {
		return nil, roachpb.NewErrorf("ID %d is not a table", id)
	}
	return tableDesc, nil
}

// resolveName resolves a table name to a descriptor ID by looking in the
// database. If the mapping is not found, errDescriptorNotFound is returned.
func (m *LeaseManager) resolveName(
	txn *client.Txn, dbID sqlbase.ID, tableName string,
) (sqlbase.ID, error) {
	nameKey := tableKey{dbID, tableName}
	key := nameKey.Key()
	gr, err := txn.Get(key)
	if err != nil {
		return 0, err
	}
	if !gr.Exists() {
		return 0, errDescriptorNotFound
	}
	return sqlbase.ID(gr.ValueInt()), nil
}

// scanCmd reads the values from the db from [key, endKey).
func scanCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	rows, pErr := txn.Scan(c.getKey(), c.getEndKey(), 0)
	if pErr != nil {
		return pErr
	}
	var vals []string
	keyPrefix := []byte(fmt.Sprintf("%d.", c.historyIdx))
	for _, kv := range rows {
		key := bytes.TrimPrefix(kv.Key, keyPrefix)
		c.env[string(key)] = kv.ValueInt()
		vals = append(vals, fmt.Sprintf("%d", kv.ValueInt()))
	}
	c.debug = fmt.Sprintf("[%s]", strings.Join(vals, " "))
	return nil
}

// commitCmd commits the transaction.
func commitCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	return txn.CommitNoCleanup()
}

// deleteCmd deletes the value at the given key from the db.
func deleteCmd(c *cmd, txn *client.Txn, t *testing.T) error {
	return txn.Del(c.getKey())
}

// commitCmd commits the transaction.
func commitCmd(c *cmd, txn *client.Txn, t *testing.T) error {
	return txn.Commit()
}

// deleteRngCmd deletes the range of values from the db from [key, endKey).
func deleteRngCmd(c *cmd, txn *client.Txn, t *testing.T) *roachpb.Error {
	return txn.DelRange(c.getKey(), c.getEndKey())
}

func (tu *tableUpdater) init(txn *client.Txn) error {
	tu.txn = txn
	tu.b = txn.NewBatch()
	return nil
}