// DecodeDataKey decodes a time series key into its components.
func DecodeDataKey(key roachpb.Key) (string, string, Resolution, int64, error) {
	// Detect and remove prefix.
	remainder := key
	if !bytes.HasPrefix(remainder, keyDataPrefix) {
		return "", "", 0, 0, util.Errorf("malformed time series data key %v: improper prefix", key)
	}
	remainder = remainder[len(keyDataPrefix):]

	// Decode series name.
	remainder, name, err := encoding.DecodeBytes(remainder, nil)
	if err != nil {
		return "", "", 0, 0, err
	}
	// Decode resolution.
	remainder, resolutionInt, err := encoding.DecodeVarint(remainder)
	if err != nil {
		return "", "", 0, 0, err
	}
	resolution := Resolution(resolutionInt)
	// Decode timestamp.
	remainder, timeslot, err := encoding.DecodeVarint(remainder)
	if err != nil {
		return "", "", 0, 0, err
	}
	timestamp := timeslot * resolution.KeyDuration()
	// The remaining bytes are the source.
	source := remainder

	return string(name), string(source), resolution, timestamp, nil
}

// DecodeDataKey decodes a time series key into its components.
func DecodeDataKey(key proto.Key) (string, string, Resolution, int64) {
	var (
		name          []byte
		source        []byte
		resolutionInt int64
		timeslot      int64
		remainder     = key
	)

	// Detect and remove prefix.
	if !bytes.HasPrefix(remainder, keyDataPrefix) {
		panic(fmt.Sprintf("malformed time series data key %v: improper prefix", key))
	}
	remainder = remainder[len(keyDataPrefix):]

	// Decode series name.
	remainder, name = encoding.DecodeBytes(remainder, nil)
	// Decode resolution.
	remainder, resolutionInt = encoding.DecodeVarint(remainder)
	resolution := Resolution(resolutionInt)
	// Decode timestamp.
	remainder, timeslot = encoding.DecodeVarint(remainder)
	timestamp := timeslot * resolution.KeyDuration()
	// The remaining bytes are the source.
	source = remainder

	return string(name), string(source), resolution, timestamp
}

func (rc *ResponseCache) decodeResponseCacheKey(encKey engine.MVCCKey) ([]byte, error) {
	key, _, isValue, err := engine.MVCCDecodeKey(encKey)
	if err != nil {
		return nil, err
	}
	if isValue {
		return nil, util.Errorf("key %s is not a raw MVCC value", encKey)
	}
	if !bytes.HasPrefix(key, keys.LocalRangeIDPrefix) {
		return nil, util.Errorf("key %s does not have %s prefix", key, keys.LocalRangeIDPrefix)
	}
	// Cut the prefix and the Range ID.
	b := key[len(keys.LocalRangeIDPrefix):]
	b, _, err = encoding.DecodeUvarint(b)
	if err != nil {
		return nil, err
	}
	if !bytes.HasPrefix(b, keys.LocalResponseCacheSuffix) {
		return nil, util.Errorf("key %s does not contain the response cache suffix %s",
			key, keys.LocalResponseCacheSuffix)
	}
	// Cut the response cache suffix.
	b = b[len(keys.LocalResponseCacheSuffix):]
	// Decode the family.
	b, fm, err := encoding.DecodeBytes(b, nil)
	if err != nil {
		return nil, err
	}
	if len(b) > 0 {
		return nil, util.Errorf("key %s has leftover bytes after decode: %s; indicates corrupt key",
			encKey, b)
	}
	return fm, nil
}

// decodeTableKey decodes a single element of a table key from b, returning the
// remaining (not yet decoded) bytes.
func decodeTableKey(b []byte, v reflect.Value) ([]byte, error) {
	switch t := v.Addr().Interface().(type) {
	case *[]byte:
		b, *t = roachencoding.DecodeBytes(b, nil)
		return b, nil
	case *string:
		var r []byte
		b, r = roachencoding.DecodeBytes(b, nil)
		*t = string(r)
		return b, nil
	}

	switch v.Kind() {
	case reflect.Bool:
		var i int64
		b, i = roachencoding.DecodeVarint(b)
		v.SetBool(i != 0)
		return b, nil

	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
		var i int64
		b, i = roachencoding.DecodeVarint(b)
		v.SetInt(i)
		return b, nil

	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr:
		var i uint64
		b, i = roachencoding.DecodeUvarint(b)
		v.SetUint(i)
		return b, nil

	case reflect.Float32, reflect.Float64:
		var f float64
		b, f = roachencoding.DecodeNumericFloat(b)
		v.SetFloat(f)
		return b, nil

	case reflect.String:
		var r []byte
		b, r = roachencoding.DecodeBytes(b, nil)
		v.SetString(string(r))
		return b, nil
	}

	return nil, fmt.Errorf("unable to decode key: %s", v)
}

// KeyAddress returns the address for the key, used to lookup the
// range containing the key. In the normal case, this is simply the
// key's value. However, for local keys, such as transaction records,
// range-spanning binary tree node pointers, and message queues, the
// address is the trailing suffix of the key, with the local key
// prefix removed. In this way, local keys address to the same range
// as non-local keys, but are stored separately so that they don't
// collide with user-space or global system keys.
//
// However, not all local keys are addressable in the global map. Only
// range local keys incorporating a range key (start key or transaction
// key) are addressable (e.g. range metadata and txn records). Range
// local keys incorporating the Raft ID are not (e.g. response cache
// entries, and range stats).
func KeyAddress(k proto.Key) proto.Key {
	if !bytes.HasPrefix(k, KeyLocalPrefix) {
		return k
	}
	if bytes.HasPrefix(k, KeyLocalRangeKeyPrefix) {
		k = k[len(KeyLocalRangeKeyPrefix):]
		_, k = encoding.DecodeBytes(k)
		return k
	}
	log.Fatalf("local key %q malformed; should contain prefix %q", k, KeyLocalRangeKeyPrefix)
	return nil
}

// DecodeRangeKey decodes the range key into range start key,
// suffix and optional detail (may be nil).
func DecodeRangeKey(key proto.Key) (startKey, suffix, detail proto.Key) {
	if !bytes.HasPrefix(key, KeyLocalRangeKeyPrefix) {
		panic(fmt.Sprintf("key %q does not have %q prefix", key, KeyLocalRangeKeyPrefix))
	}
	// Cut the prefix and the Raft ID.
	b := key[len(KeyLocalRangeKeyPrefix):]
	b, startKey = encoding.DecodeBytes(b)
	if len(b) < KeyLocalSuffixLength {
		panic(fmt.Sprintf("key %q does not have suffix of length %d", key, KeyLocalSuffixLength))
	}
	// Cut the response cache suffix.
	suffix = b[:KeyLocalSuffixLength]
	detail = b[KeyLocalSuffixLength:]
	return
}

// ShowDatabases returns all the databases.
// Privileges: None.
//   Notes: postgres does not have a "show databases"
//          mysql has a "SHOW DATABASES" permission, but we have no system-level permissions.
func (p *planner) ShowDatabases(n *parser.ShowDatabases) (planNode, error) {
	// TODO(pmattis): This could be implemented as:
	//
	//   SELECT id FROM system.namespace WHERE parentID = 0

	prefix := MakeNameMetadataKey(RootNamespaceID, "")
	sr, err := p.txn.Scan(prefix, prefix.PrefixEnd(), 0)
	if err != nil {
		return nil, err
	}
	v := &valuesNode{columns: []string{"Database"}}
	for _, row := range sr {
		_, name := encoding.DecodeBytes(bytes.TrimPrefix(row.Key, prefix), nil)
		v.rows = append(v.rows, []parser.Datum{parser.DString(name)})
	}
	return v, nil
}

func decodeIndexKey(desc *structured.TableDescriptor,
	index structured.IndexDescriptor, vals map[string]driver.Value, key []byte) ([]byte, error) {
	if !bytes.HasPrefix(key, keys.TableDataPrefix) {
		return nil, fmt.Errorf("%s: invalid key prefix: %q", desc.Name, key)
	}
	key = bytes.TrimPrefix(key, keys.TableDataPrefix)

	var tableID uint64
	key, tableID = encoding.DecodeUvarint(key)
	if uint32(tableID) != desc.ID {
		return nil, fmt.Errorf("%s: unexpected table ID: %d != %d", desc.Name, desc.ID, tableID)
	}

	var indexID uint64
	key, indexID = encoding.DecodeUvarint(key)
	if uint32(indexID) != index.ID {
		return nil, fmt.Errorf("%s: unexpected index ID: %d != %d", desc.Name, index.ID, indexID)
	}

	for _, id := range index.ColumnIDs {
		col, err := findColumnByID(desc, id)
		if err != nil {
			return nil, err
		}
		switch col.Type.Kind {
		case structured.ColumnType_BIT, structured.ColumnType_INT:
			var i int64
			key, i = encoding.DecodeVarint(key)
			vals[col.Name] = i
		case structured.ColumnType_FLOAT:
			var f float64
			key, f = encoding.DecodeNumericFloat(key)
			vals[col.Name] = f
		case structured.ColumnType_CHAR, structured.ColumnType_BINARY,
			structured.ColumnType_TEXT, structured.ColumnType_BLOB:
			var r []byte
			key, r = encoding.DecodeBytes(key, nil)
			vals[col.Name] = r
		default:
			return nil, fmt.Errorf("TODO(pmattis): decoded index key: %s", col.Type.Kind)
		}
	}

	return key, nil
}

// Addr returns the address for the key, used to lookup the range containing
// the key. In the normal case, this is simply the key's value. However, for
// local keys, such as transaction records, range-spanning binary tree node
// pointers, the address is the trailing suffix of the key, with the local key
// prefix removed. In this way, local keys address to the same range as
// non-local keys, but are stored separately so that they don't collide with
// user-space or global system keys.
//
// However, not all local keys are addressable in the global map. Only range
// local keys incorporating a range key (start key or transaction key) are
// addressable (e.g. range metadata and txn records). Range local keys
// incorporating the Range ID are not (e.g. response cache entries, and range
// stats).
//
// TODO(pmattis): Should KeyAddress return an error when the key is malformed?
func Addr(k roachpb.Key) roachpb.RKey {
	if k == nil {
		return nil
	}

	if !bytes.HasPrefix(k, localPrefix) {
		return roachpb.RKey(k)
	}
	if bytes.HasPrefix(k, LocalRangePrefix) {
		k = k[len(LocalRangePrefix):]
		_, k, err := encoding.DecodeBytes(k, nil)
		if err != nil {
			panic(err)
		}
		return roachpb.RKey(k)
	}
	log.Fatalf("local key %q malformed; should contain prefix %q",
		k, LocalRangePrefix)
	return nil
}

// DecodeRangeKey decodes the range key into range start key,
// suffix and optional detail (may be nil).
func DecodeRangeKey(key roachpb.Key) (startKey, suffix, detail roachpb.Key, err error) {
	if !bytes.HasPrefix(key, LocalRangePrefix) {
		return nil, nil, nil, util.Errorf("key %q does not have %q prefix",
			key, LocalRangePrefix)
	}
	// Cut the prefix and the Range ID.
	b := key[len(LocalRangePrefix):]
	b, startKey, err = encoding.DecodeBytes(b, nil)
	if err != nil {
		return nil, nil, nil, err
	}
	if len(b) < localSuffixLength {
		return nil, nil, nil, util.Errorf("key %q does not have suffix of length %d",
			key, localSuffixLength)
	}
	// Cut the response cache suffix.
	suffix = b[:localSuffixLength]
	detail = b[localSuffixLength:]
	return
}

// KeyAddress returns the address for the key, used to lookup the
// range containing the key. In the normal case, this is simply the
// key's value. However, for local keys, such as transaction records,
// range-spanning binary tree node pointers, and message queues, the
// address is the trailing suffix of the key, with the local key
// prefix removed. In this way, local keys address to the same range
// as non-local keys, but are stored separately so that they don't
// collide with user-space or global system keys.
//
// However, not all local keys are addressable in the global map. Only
// range local keys incorporating a range key (start key or transaction
// key) are addressable (e.g. range metadata and txn records). Range
// local keys incorporating the Range ID are not (e.g. response cache
// entries, and range stats).
//
// TODO(pmattis): Should KeyAddress return an error when the key is malformed?
func KeyAddress(k proto.Key) proto.Key {
	if k == nil {
		return nil
	}

	if !bytes.HasPrefix(k, LocalPrefix) {
		return k
	}
	if bytes.HasPrefix(k, LocalRangePrefix) {
		k = k[len(LocalRangePrefix):]
		_, k, err := encoding.DecodeBytes(k, nil)
		if err != nil {
			panic(err)
		}
		return k
	}
	log.Fatalf("local key %q malformed; should contain prefix %q",
		k, LocalRangePrefix)
	return nil
}

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

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

// decodeIndexKey decodes the values that are a part of the specified index
// key. Vals is a slice returned from makeIndexKeyVals. The remaining bytes in
// the index key are returned which will either be an encoded column ID for the
// primary key index, the primary key suffix for non-unique secondary indexes
// or unique secondary indexes containing NULL or empty.
func decodeIndexKey(desc *structured.TableDescriptor,
	index structured.IndexDescriptor, vals []parser.Datum, key []byte) ([]byte, error) {
	if !bytes.HasPrefix(key, keys.TableDataPrefix) {
		return nil, fmt.Errorf("%s: invalid key prefix: %q", desc.Name, key)
	}
	key = bytes.TrimPrefix(key, keys.TableDataPrefix)

	var tableID uint64
	key, tableID = encoding.DecodeUvarint(key)
	if structured.ID(tableID) != desc.ID {
		return nil, fmt.Errorf("%s: unexpected table ID: %d != %d", desc.Name, desc.ID, tableID)
	}

	var indexID uint64
	key, indexID = encoding.DecodeUvarint(key)
	if structured.IndexID(indexID) != index.ID {
		return nil, fmt.Errorf("%s: unexpected index ID: %d != %d", desc.Name, index.ID, indexID)
	}

	for j := range vals {
		switch vals[j].(type) {
		case parser.DInt:
			var i int64
			key, i = encoding.DecodeVarint(key)
			vals[j] = parser.DInt(i)
		case parser.DFloat:
			var f float64
			key, f = encoding.DecodeNumericFloat(key)
			vals[j] = parser.DFloat(f)
		case parser.DString:
			var r []byte
			key, r = encoding.DecodeBytes(key, nil)
			vals[j] = parser.DString(r)
		default:
			return nil, util.Errorf("TODO(pmattis): decoded index key: %s", vals[j].Type())
		}
	}

	return key, nil
}

func sequenceCacheKeyPrint(key roachpb.Key) string {
	b, id, err := encoding.DecodeBytes([]byte(key), nil)
	if err != nil {
		return fmt.Sprintf("/%q/err:%v", key, err)
	}

	if len(b) == 0 {
		return fmt.Sprintf("/%q", id)
	}

	b, epoch, err := encoding.DecodeUint32Decreasing(b)
	if err != nil {
		return fmt.Sprintf("/%q/err:%v", id, err)
	}

	_, seq, err := encoding.DecodeUint32Decreasing(b)
	if err != nil {
		return fmt.Sprintf("/%q/epoch:%d/err:%v", id, epoch, err)
	}

	return fmt.Sprintf("/%q/epoch:%d/seq:%d", id, epoch, seq)
}

// decodeKeyVals decodes the values that are part of the key. Vals is a slice
// returned from makeKeyVals. The remaining bytes in the key after decoding the
// values are returned.
func decodeKeyVals(vals []parser.Datum, key []byte) ([]byte, error) {
	for j := range vals {
		switch vals[j].(type) {
		case parser.DInt:
			var i int64
			key, i = encoding.DecodeVarint(key)
			vals[j] = parser.DInt(i)
		case parser.DFloat:
			var f float64
			key, f = encoding.DecodeNumericFloat(key)
			vals[j] = parser.DFloat(f)
		case parser.DString:
			var r []byte
			key, r = encoding.DecodeBytes(key, nil)
			vals[j] = parser.DString(r)
		default:
			return nil, util.Errorf("TODO(pmattis): decoded index key: %s", vals[j].Type())
		}
	}

	return key, nil
}

func decodeSequenceCacheKey(key roachpb.Key, dest []byte) ([]byte, uint32, uint32, error) {
	// TODO(tschottdorf): redundant check.
	if !bytes.HasPrefix(key, keys.LocalRangeIDPrefix) {
		return nil, 0, 0, util.Errorf("key %s does not have %s prefix", key, keys.LocalRangeIDPrefix)
	}
	// Cut the prefix and the Range ID.
	b := key[len(keys.LocalRangeIDPrefix):]
	b, _, err := encoding.DecodeUvarint(b)
	if err != nil {
		return nil, 0, 0, err
	}
	if !bytes.HasPrefix(b, keys.LocalSequenceCacheSuffix) {
		return nil, 0, 0, util.Errorf("key %s does not contain the sequence cache suffix %s",
			key, keys.LocalSequenceCacheSuffix)
	}
	// Cut the sequence cache suffix.
	b = b[len(keys.LocalSequenceCacheSuffix):]
	// Decode the id.
	b, id, err := encoding.DecodeBytes(b, dest)
	if err != nil {
		return nil, 0, 0, err
	}
	// Decode the epoch.
	b, epoch, err := encoding.DecodeUint32Decreasing(b)
	if err != nil {
		return nil, 0, 0, err
	}
	// Decode the sequence number.
	b, seq, err := encoding.DecodeUint32Decreasing(b)
	if err != nil {
		return nil, 0, 0, err
	}
	if len(b) > 0 {
		return nil, 0, 0, util.Errorf("key %q has leftover bytes after decode: %s; indicates corrupt key",
			key, b)
	}
	return id, epoch, seq, nil
}