func (rc *ResponseCache) decodeResponseCacheKey(encKey proto.EncodedKey) (proto.ClientCmdID, error) {
	ret := proto.ClientCmdID{}
	key, _, isValue := engine.MVCCDecodeKey(encKey)
	if isValue {
		return ret, util.Errorf("key %s is not a raw MVCC value", encKey)
	}
	if !bytes.HasPrefix(key, keys.LocalRangeIDPrefix) {
		return ret, util.Errorf("key %s does not have %s prefix", key, keys.LocalRangeIDPrefix)
	}
	// Cut the prefix and the Raft ID.
	b := key[len(keys.LocalRangeIDPrefix):]
	b, _ = encoding.DecodeUvarint(b)
	if !bytes.HasPrefix(b, keys.LocalResponseCacheSuffix) {
		return ret, 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):]
	// Now, decode the command ID.
	b, wt := encoding.DecodeUvarint(b)
	b, rd := encoding.DecodeUint64(b)
	if len(b) > 0 {
		return ret, util.Errorf("key %s has leftover bytes after decode: %s; indicates corrupt key",
			encKey, b)
	}
	ret.WallTime = int64(wt)
	ret.Random = int64(rd)
	return ret, nil
}

// GetLargestObjectID returns the largest object ID found in the config.
// This could be either a table or a database.
func (s *SystemConfig) GetLargestObjectID() (uint32, error) {
	testingLock.Lock()
	hook := TestingLargestIDHook
	testingLock.Unlock()
	if hook != nil {
		return hook(), nil
	}

	if len(s.Values) == 0 {
		return 0, fmt.Errorf("empty system values in config")
	}

	// Search for the first key after the descriptor table.
	// We can't use GetValue as we don't mind if there is nothing after
	// the descriptor table.
	key := proto.Key(keys.MakeTablePrefix(keys.DescriptorTableID + 1))
	index := sort.Search(len(s.Values), func(i int) bool {
		return !s.Values[i].Key.Less(key)
	})

	if index == 0 {
		return 0, fmt.Errorf("descriptor table not found in system config of %d values", len(s.Values))
	}

	// This is the last key of the descriptor table.
	key = s.Values[index-1].Key

	// Extract object ID from key.
	// TODO(marc): move sql/keys.go to keys (or similar) and use a DecodeDescMetadataKey.
	// We should also check proper encoding.
	descriptorPrefix := keys.MakeTablePrefix(keys.DescriptorTableID)
	remaining := bytes.TrimPrefix(key, descriptorPrefix)
	// TrimPrefix returns the bytes unchanged if the prefix does not match.
	if len(remaining) == len(key) {
		return 0, fmt.Errorf("descriptor table not found in system config of %d values", len(s.Values))
	}
	// DescriptorTable.PrimaryIndex.ID
	remaining, _, err := encoding.DecodeUvarint(remaining)
	if err != nil {
		return 0, err
	}
	// descID
	_, id, err := encoding.DecodeUvarint(remaining)
	if err != nil {
		return 0, err
	}
	return uint32(id), nil
}

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
}

// MakeSplitKey transforms an SQL table key such that it is a valid split key
// (i.e. does not occur in the middle of a row).
func MakeSplitKey(key roachpb.Key) (roachpb.Key, error) {
	if encoding.PeekType(key) != encoding.Int {
		// Not a table key, so already a split key.
		return key, nil
	}

	n := len(key)
	// The column ID length is encoded as a varint and we take advantage of the
	// fact that the column ID itself will be encoded in 0-9 bytes and thus the
	// length of the column ID data will fit in a single byte.
	buf := key[n-1:]
	if encoding.PeekType(buf) != encoding.Int {
		// The last byte is not a valid column ID suffix.
		return nil, util.Errorf("%s: not a valid table key", key)
	}

	// Strip off the column ID suffix from the buf. The last byte of the buf
	// contains the length of the column ID suffix (which might be 0 if the buf
	// does not contain a column ID suffix).
	_, colIDLen, err := encoding.DecodeUvarint(buf)
	if err != nil {
		return nil, err
	}
	if int(colIDLen)+1 > n {
		// The column ID length was impossible. colIDLen is the length of the
		// encoded column ID suffix. We add 1 to account for the byte holding the
		// length of the encoded column ID and if that total (colIDLen+1) is
		// greater than the key suffix (n == len(buf)) then we bail. Note that we
		// don't consider this an error because MakeSplitKey can be called on keys
		// that look like table keys but which do not have a column ID length
		// suffix (e.g SystemConfig.ComputeSplitKeys).
		return nil, util.Errorf("%s: malformed table key", key)
	}
	return key[:len(key)-int(colIDLen)-1], nil
}

func decodeIndexKeyPrefix(desc *TableDescriptor, key []byte) (IndexID, []byte, error) {
	var tableID, indexID uint64

	if !bytes.HasPrefix(key, keys.TableDataPrefix) {
		return IndexID(indexID), nil, util.Errorf("%s: invalid key prefix: %q", desc.Name, key)
	}

	key = bytes.TrimPrefix(key, keys.TableDataPrefix)
	key, tableID = encoding.DecodeUvarint(key)
	key, indexID = encoding.DecodeUvarint(key)

	if ID(tableID) != desc.ID {
		return IndexID(indexID), nil, util.Errorf("%s: unexpected table ID: %d != %d", desc.Name, desc.ID, tableID)
	}

	return IndexID(indexID), key, 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
}

// DecodeRaftStateKey extracts the Range ID from a RaftStateKey.
func DecodeRaftStateKey(key proto.Key) proto.RangeID {
	if !bytes.HasPrefix(key, LocalRangeIDPrefix) {
		panic(fmt.Sprintf("key %q does not have %q prefix", key, LocalRangeIDPrefix))
	}
	// Cut the prefix and the Range ID.
	b := key[len(LocalRangeIDPrefix):]
	_, rangeID := encoding.DecodeUvarint(b)
	return proto.RangeID(rangeID)
}

func decodeIndexKeyPrefix(desc *TableDescriptor, key []byte) (IndexID, []byte, *roachpb.Error) {
	if encoding.PeekType(key) != encoding.Int {
		return 0, nil, roachpb.NewErrorf("%s: invalid key prefix: %q", desc.Name, key)
	}

	key, tableID, err := encoding.DecodeUvarint(key)
	if err != nil {
		return 0, nil, roachpb.NewError(err)
	}
	key, indexID, err := encoding.DecodeUvarint(key)
	if err != nil {
		return 0, nil, roachpb.NewError(err)
	}

	if ID(tableID) != desc.ID {
		return IndexID(indexID), nil, roachpb.NewErrorf("%s: unexpected table ID: %d != %d", desc.Name, desc.ID, tableID)
	}

	return IndexID(indexID), key, nil
}

// decodeIndexKey decodes the values that are a part of the specified index
// key. ValTypes is a slice returned from makeKeyVals. 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 *TableDescriptor,
	index IndexDescriptor, valTypes, 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 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 IndexID(indexID) != index.ID {
		return nil, fmt.Errorf("%s: unexpected index ID: %d != %d", desc.Name, index.ID, indexID)
	}

	return decodeKeyVals(valTypes, vals, key)
}

// ObjectIDForKey returns the object ID (table or database) for 'key',
// or (_, false) if not within the structured key space.
func ObjectIDForKey(key roachpb.RKey) (uint32, bool) {
	if key.Equal(roachpb.RKeyMax) {
		return 0, false
	}
	if encoding.PeekType(key) != encoding.Int {
		// TODO(marc): this should eventually return SystemDatabaseID.
		return 0, false
	}
	// Consume first encoded int.
	_, id64, err := encoding.DecodeUvarint(key)
	return uint32(id64), err == 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 decodeDescMetadataID(key roachpb.Key) (uint64, error) {
	// Extract object ID from key.
	// TODO(marc): move sql/keys.go to keys (or similar) and use a DecodeDescMetadataKey.
	// We should also check proper encoding.
	remaining, tableID, err := keys.DecodeTablePrefix(key)
	if err != nil {
		return 0, err
	}
	if tableID != keys.DescriptorTableID {
		return 0, util.Errorf("key is not a descriptor table entry: %v", key)
	}
	// DescriptorTable.PrimaryIndex.ID
	remaining, _, err = encoding.DecodeUvarint(remaining)
	if err != nil {
		return 0, err
	}
	// descID
	_, id, err := encoding.DecodeUvarint(remaining)
	if err != nil {
		return 0, err
	}
	return id, 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)
}

// ObjectIDForKey returns the object ID (table or database) for 'key',
// or (_, false) if not within the structured key space.
func ObjectIDForKey(key proto.Key) (uint32, bool) {
	if key.Equal(proto.KeyMax) {
		return 0, false
	}
	if key.Equal(keys.TableDataPrefix) {
		// TODO(marc): this should eventually return SystemDatabaseID.
		return 0, false
	}
	remaining := bytes.TrimPrefix(key, keys.TableDataPrefix)
	if len(remaining) == len(key) {
		// TrimPrefix returns the input untouched if the prefix doesn't match.
		return 0, false
	}

	// Consume first encoded int.
	_, id64, err := encoding.DecodeUvarint(remaining)
	return uint32(id64), err == nil
}

// decodePrimaryKey decodes a primary key for the table into the model object
// v. It returns the remaining (undecoded) bytes.
func (m *model) decodePrimaryKey(key []byte, v reflect.Value) ([]byte, error) {
	if !bytes.HasPrefix(key, keys.TableDataPrefix) {
		return nil, fmt.Errorf("%s: invalid key prefix: %q", m.name, key)
	}
	key = bytes.TrimPrefix(key, keys.TableDataPrefix)

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

	for _, col := range m.primaryKey {
		var err error
		key, err = decodeTableKey(key, v.FieldByIndex(col.field.Index))
		if err != nil {
			return nil, err
		}
	}

	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
}

// ObjectIDForKey returns the object ID (table or database) for 'key',
// or (_, false) if not within the structured key space.
func ObjectIDForKey(key proto.Key) (uint32, bool) {
	if key.Equal(proto.KeyMax) {
		return 0, false
	}
	if key.Equal(keys.TableDataPrefix) {
		// TODO(marc): this should eventually return SystemDatabaseID.
		return 0, false
	}
	remaining := bytes.TrimPrefix(key, keys.TableDataPrefix)
	if len(remaining) == len(key) {
		// TrimPrefix returns the input untouched if the prefix doesn't match.
		return 0, false
	}

	// Consume first encoded int.
	defer func() {
		// Nothing to do, default return values mean "could not decode", which is
		// definitely the case if DecodeUvarint panics.
		_ = recover()
	}()
	_, id64 := encoding.DecodeUvarint(remaining)
	return uint32(id64), true
}

//.........这里部分代码省略.........
		lowerStrings(columns)
		scanColIDs = make(map[uint32]bool, len(columns))
		for _, colName := range columns {
			col, ok := m.columnsByName[colName]
			if !ok {
				b.initResult(0, 0, fmt.Errorf("%s: unable to find column %s", m.name, colName))
				return
			}
			scanColIDs[col.ID] = true
		}
	}

	startV := reflect.Indirect(reflect.ValueOf(start))
	if modelT != startV.Type() {
		b.initResult(0, 0, fmt.Errorf("incompatible start key type: %s != %s", modelT, startV.Type()))
		return
	}

	endV := reflect.Indirect(reflect.ValueOf(end))
	if modelT != endV.Type() {
		b.initResult(0, 0, fmt.Errorf("incompatible end key type: %s != %s", modelT, endV.Type()))
		return
	}

	startKey, err := m.encodePrimaryKey(startV)
	if err != nil {
		b.initResult(0, 0, err)
		return
	}
	endKey, err := m.encodePrimaryKey(endV)
	if err != nil {
		b.initResult(0, 0, err)
		return
	}
	if log.V(2) {
		log.Infof("Scan %q %q", startKey, endKey)
	}

	c := proto.ScanCall(proto.Key(startKey), proto.Key(endKey), maxRows)
	c.Post = func() error {
		reply := c.Reply.(*proto.ScanResponse)
		if len(reply.Rows) == 0 {
			return nil
		}

		var primaryKey []byte
		resultPtr := reflect.New(modelT)
		result := resultPtr.Elem()
		zero := reflect.Zero(result.Type())

		for _, row := range reply.Rows {
			if primaryKey != nil && !bytes.HasPrefix(row.Key, primaryKey) {
				if ptrResults {
					sliceV = reflect.Append(sliceV, resultPtr)
					resultPtr = reflect.New(modelT)
					result = resultPtr.Elem()
				} else {
					sliceV = reflect.Append(sliceV, result)
					result.Set(zero)
				}
				_, err := m.decodePrimaryKey(primaryKey, result)
				if err != nil {
					return err
				}
			}

			remaining, err := m.decodePrimaryKey([]byte(row.Key), result)
			if err != nil {
				return err
			}
			primaryKey = []byte(row.Key[:len(row.Key)-len(remaining)])

			_, colID := roachencoding.DecodeUvarint(remaining)
			if err != nil {
				return err
			}
			if scanColIDs != nil && !scanColIDs[uint32(colID)] {
				continue
			}
			col, ok := m.columnsByID[uint32(colID)]
			if !ok {
				return fmt.Errorf("%s: unable to find column %d", m.name, colID)
			}
			if err := unmarshalValue(&row.Value, result.FieldByIndex(col.field.Index)); err != nil {
				return err
			}
		}

		if ptrResults {
			sliceV = reflect.Append(sliceV, resultPtr)
		} else {
			sliceV = reflect.Append(sliceV, result)
		}
		reflect.ValueOf(dest).Elem().Set(sliceV)
		return nil
	}

	b.calls = append(b.calls, c)
	b.initResult(1, 0, nil)
}

func (c *conn) Select(p *parser.Select, args []driver.Value) (*rows, error) {
	if len(p.Exprs) != 1 {
		return nil, fmt.Errorf("TODO(pmattis): unsupported select exprs: %s", p.Exprs)
	}
	if _, ok := p.Exprs[0].(*parser.StarExpr); !ok {
		return nil, fmt.Errorf("TODO(pmattis): unsupported select expr: %s", p.Exprs)
	}

	if len(p.From) != 1 {
		return nil, fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
	}
	var desc *structured.TableDescriptor
	{
		ate, ok := p.From[0].(*parser.AliasedTableExpr)
		if !ok {
			return nil, fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
		}
		table, ok := ate.Expr.(*parser.TableName)
		if !ok {
			return nil, fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
		}
		var err error
		desc, err = c.getTableDesc(table)
		if err != nil {
			return nil, err
		}
	}

	// Retrieve all of the keys that start with our index key prefix.
	startKey := proto.Key(encodeIndexKeyPrefix(desc.ID, desc.Indexes[0].ID))
	endKey := startKey.PrefixEnd()
	sr, err := c.db.Scan(startKey, endKey, 0)
	if err != nil {
		return nil, err
	}

	// All of the columns for a particular row will be grouped together. We loop
	// over the returned key/value pairs and decode the key to extract the
	// columns encoded within the key and the column ID. We use the column ID to
	// lookup the column and decode the value. All of these values go into a map
	// keyed by column name. When the index key changes we output a row
	// containing the current values.
	//
	// The TODOs here are too numerous to list. This is only performing a full
	// table scan using the primary key.

	r := &rows{}
	var primaryKey []byte
	vals := map[string]driver.Value{}
	for _, kv := range sr {
		if primaryKey != nil && !bytes.HasPrefix(kv.Key, primaryKey) {
			outputRow(r, desc.Columns, vals)
			vals = map[string]driver.Value{}
		}

		remaining, err := decodeIndexKey(desc, desc.Indexes[0], vals, kv.Key)
		if err != nil {
			return nil, err
		}
		primaryKey = []byte(kv.Key[:len(kv.Key)-len(remaining)])

		_, colID := encoding.DecodeUvarint(remaining)
		if err != nil {
			return nil, err
		}
		col, err := findColumnByID(desc, uint32(colID))
		if err != nil {
			return nil, err
		}
		vals[col.Name] = unmarshalValue(col, kv)

		if log.V(2) {
			log.Infof("Scan %q -> %v", kv.Key, vals[col.Name])
		}
	}

	outputRow(r, desc.Columns, vals)

	r.columns = make([]string, len(desc.Columns))
	for i, col := range desc.Columns {
		r.columns[i] = col.Name
	}
	return r, nil
}

// Select selects rows from a single table.
func (s *Server) Select(session *Session, p *parser.Select, args []sqlwire.Datum, resp *sqlwire.Response) error {
	if len(p.Exprs) != 1 {
		return fmt.Errorf("TODO(pmattis): unsupported select exprs: %s", p.Exprs)
	}
	if _, ok := p.Exprs[0].(*parser.StarExpr); !ok {
		return fmt.Errorf("TODO(pmattis): unsupported select expr: %s", p.Exprs)
	}

	if len(p.From) != 1 {
		return fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
	}
	var desc *structured.TableDescriptor
	{
		ate, ok := p.From[0].(*parser.AliasedTableExpr)
		if !ok {
			return fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
		}
		table, ok := ate.Expr.(parser.QualifiedName)
		if !ok {
			return fmt.Errorf("TODO(pmattis): unsupported from: %s", p.From)
		}
		var err error
		desc, err = s.getTableDesc(session.Database, table)
		if err != nil {
			return err
		}
	}

	// Retrieve all of the keys that start with our index key prefix.
	startKey := proto.Key(encodeIndexKeyPrefix(desc.ID, desc.Indexes[0].ID))
	endKey := startKey.PrefixEnd()
	sr, err := s.db.Scan(startKey, endKey, 0)
	if err != nil {
		return err
	}

	// All of the columns for a particular row will be grouped together. We loop
	// over the returned key/value pairs and decode the key to extract the
	// columns encoded within the key and the column ID. We use the column ID to
	// lookup the column and decode the value. All of these values go into a map
	// keyed by column name. When the index key changes we output a row
	// containing the current values.
	//
	// The TODOs here are too numerous to list. This is only performing a full
	// table scan using the primary key.

	var rows []sqlwire.Result_Row
	var primaryKey []byte
	vals := valMap{}
	for _, kv := range sr {
		if primaryKey != nil && !bytes.HasPrefix(kv.Key, primaryKey) {
			if output, err := shouldOutputRow(p.Where, vals); err != nil {
				return err
			} else if output {
				rows = append(rows, outputRow(desc.Columns, vals))
			}
			vals = valMap{}
		}

		remaining, err := decodeIndexKey(desc, desc.Indexes[0], vals, kv.Key)
		if err != nil {
			return err
		}
		primaryKey = []byte(kv.Key[:len(kv.Key)-len(remaining)])

		_, colID := encoding.DecodeUvarint(remaining)
		if err != nil {
			return err
		}
		col, err := desc.FindColumnByID(uint32(colID))
		if err != nil {
			return err
		}
		vals[col.Name] = unmarshalValue(*col, kv)

		if log.V(2) {
			log.Infof("Scan %q -> %v", kv.Key, vals[col.Name])
		}
	}

	if output, err := shouldOutputRow(p.Where, vals); err != nil {
		return err
	} else if output {
		rows = append(rows, outputRow(desc.Columns, vals))
	}

	resp.Results = []sqlwire.Result{
		{
			Columns: make([]string, len(desc.Columns)),
			Rows:    rows,
		},
	}
	for i, col := range desc.Columns {
		resp.Results[0].Columns[i] = col.Name
	}
	return nil
}