// DecodeTableIDIndexID decodes a table id followed by an index id.
func DecodeTableIDIndexID(key []byte) ([]byte, ID, IndexID, error) {
	var tableID uint64
	var indexID uint64
	var err error

	key, tableID, err = encoding.DecodeUvarintAscending(key)
	if err != nil {
		return nil, 0, 0, err
	}
	key, indexID, err = encoding.DecodeUvarintAscending(key)
	if err != nil {
		return nil, 0, 0, err
	}

	return key, ID(tableID), IndexID(indexID), 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, errors.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.DecodeUvarintAscending(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, errors.Errorf("%s: malformed table key", key)
	}
	return key[:len(key)-int(colIDLen)-1], nil
}

// DecodeAbortCacheKey decodes the provided abort cache entry,
// returning the transaction ID.
func DecodeAbortCacheKey(key roachpb.Key, dest []byte) (*uuid.UUID, error) {
	// TODO(tschottdorf): redundant check.
	if !bytes.HasPrefix(key, LocalRangeIDPrefix) {
		return nil, util.Errorf("key %s does not have %s prefix", key, LocalRangeIDPrefix)
	}
	// Cut the prefix, the Range ID, and the infix specifier.
	b := key[len(LocalRangeIDPrefix):]
	b, _, err := encoding.DecodeUvarintAscending(b)
	if err != nil {
		return nil, err
	}
	b = b[1:]
	if !bytes.HasPrefix(b, LocalAbortCacheSuffix) {
		return nil, util.Errorf("key %s does not contain the abort cache suffix %s",
			key, LocalAbortCacheSuffix)
	}
	// Cut the abort cache suffix.
	b = b[len(LocalAbortCacheSuffix):]
	// Decode the id.
	b, idBytes, err := encoding.DecodeBytesAscending(b, dest)
	if err != nil {
		return nil, err
	}
	if len(b) > 0 {
		return nil, util.Errorf("key %q has leftover bytes after decode: %s; indicates corrupt key", key, b)
	}
	txnID, err := uuid.FromBytes(idBytes)
	return txnID, err
}

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

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

	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
}

// 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.DecodeUvarintAscending(key)
	return uint32(id64), err == 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.DecodeUvarintAscending(remaining)
	if err != nil {
		return 0, err
	}
	// descID
	_, id, err := encoding.DecodeUvarintAscending(remaining)
	if err != nil {
		return 0, err
	}
	return id, nil
}

// DecodeRangeIDKey parses a local range ID key into range ID, infix,
// suffix, and detail.
func DecodeRangeIDKey(key roachpb.Key) (rangeID roachpb.RangeID, infix, suffix, detail roachpb.Key, err error) {
	if !bytes.HasPrefix(key, LocalRangeIDPrefix) {
		return 0, nil, nil, nil, errors.Errorf("key %s does not have %s prefix", key, LocalRangeIDPrefix)
	}
	// Cut the prefix, the Range ID, and the infix specifier.
	b := key[len(LocalRangeIDPrefix):]
	b, rangeInt, err := encoding.DecodeUvarintAscending(b)
	if err != nil {
		return 0, nil, nil, nil, err
	}
	if len(b) < localSuffixLength+1 {
		return 0, nil, nil, nil, errors.Errorf("malformed key does not contain range ID infix and suffix")
	}
	infix = b[:1]
	b = b[1:]
	suffix = b[:localSuffixLength]
	b = b[localSuffixLength:]

	return roachpb.RangeID(rangeInt), infix, suffix, b, nil
}

func decodeSequenceCacheKey(key roachpb.Key, dest []byte) (*uuid.UUID, 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.DecodeUvarintAscending(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, idBytes, err := encoding.DecodeBytesAscending(b, dest)
	if err != nil {
		return nil, 0, 0, err
	}
	// Decode the epoch.
	b, epoch, err := encoding.DecodeUint32Descending(b)
	if err != nil {
		return nil, 0, 0, err
	}
	// Decode the sequence number.
	b, seq, err := encoding.DecodeUint32Descending(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)
	}
	txnID, err := uuid.FromBytes(idBytes)
	return txnID, epoch, seq, err
}

func (n *scanNode) processKV(kv client.KeyValue) bool {
	if n.indexKey == nil {
		// Reset the row to nil; it will get filled in in with the column
		// values as we decode the key-value pairs for the row.
		for i := range n.row {
			n.row[i] = nil
		}
	}

	var remaining []byte
	var err error
	remaining, err = n.readIndexKey(kv.Key)
	n.pErr = roachpb.NewError(err)
	if n.pErr != nil {
		return false
	}

	if n.indexKey == nil {
		n.indexKey = []byte(kv.Key[:len(kv.Key)-len(remaining)])

		// This is the first key for the row, initialize the column values that are
		// part of the index key.
		for i, id := range n.columnIDs {
			if idx, ok := n.colIdxMap[id]; ok {
				n.row[idx] = n.vals[i]
			}
		}
	}

	var value parser.Datum
	n.colID = 0

	if !n.isSecondaryIndex && len(remaining) > 0 {
		var v uint64
		var err error
		_, v, err = encoding.DecodeUvarintAscending(remaining)
		n.pErr = roachpb.NewError(err)
		if n.pErr != nil {
			return false
		}
		n.colID = ColumnID(v)
		if idx, ok := n.colIdxMap[n.colID]; ok && n.valNeededForCol[idx] {
			value, ok = n.unmarshalValue(kv)
			if !ok {
				return false
			}
			if n.row[idx] != nil {
				panic(fmt.Sprintf("duplicate value for column %d", idx))
			}
			n.row[idx] = value
			if log.V(2) {
				log.Infof("Scan %s -> %v", kv.Key, value)
			}
		} else {
			// No need to unmarshal the column value. Either the column was part of
			// the index key or it isn't needed.
			if log.V(2) {
				log.Infof("Scan %s -> [%d] (skipped)", kv.Key, n.colID)
			}
		}
	} else {
		if n.implicitVals != nil {
			implicitDirs := make([]encoding.Direction, 0, len(n.index.ImplicitColumnIDs))
			for range n.index.ImplicitColumnIDs {
				implicitDirs = append(implicitDirs, encoding.Ascending)
			}
			var err error
			_, err = decodeKeyVals(n.implicitValTypes, n.implicitVals, implicitDirs, kv.ValueBytes())
			n.pErr = roachpb.NewError(err)
			if n.pErr != nil {
				return false
			}
			for i, id := range n.index.ImplicitColumnIDs {
				if idx, ok := n.colIdxMap[id]; ok && n.valNeededForCol[idx] {
					n.row[idx] = n.implicitVals[i]
				}
			}
		}

		if log.V(2) {
			if n.implicitVals != nil {
				log.Infof("Scan %s -> %s", kv.Key, prettyDatums(n.implicitVals))
			} else {
				log.Infof("Scan %s", kv.Key)
			}
		}
	}

	if n.explain == explainDebug {
		if value == nil {
			if n.colID > 0 {
				var ok bool
				value, ok = n.unmarshalValue(kv)
				if !ok {
					return false
				}
			} else {
				value = parser.DNull
			}
		}
//.........这里部分代码省略.........

func (n *scanNode) processKV(kv client.KeyValue) bool {
	if n.indexKey == nil {
		// Reset the qvals map expressions to nil. The expressions will get filled
		// in with the column values as we decode the key-value pairs for the row.
		for _, qval := range n.qvals {
			qval.datum = nil
		}
	}

	var remaining []byte
	remaining, n.pErr = decodeIndexKey(n.desc, n.index.ID, n.valTypes, n.vals, n.columnDirs, kv.Key)
	if n.pErr != nil {
		return false
	}

	if n.indexKey == nil {
		n.indexKey = []byte(kv.Key[:len(kv.Key)-len(remaining)])

		// This is the first key for the row, initialize the column values that are
		// part of the index key.
		for i, id := range n.columnIDs {
			if qval, ok := n.qvals[id]; ok {
				qval.datum = n.vals[i]
			}
		}
	}

	var value parser.Datum
	n.colID = 0

	if !n.isSecondaryIndex && len(remaining) > 0 {
		var v uint64
		var err error
		_, v, err = encoding.DecodeUvarintAscending(remaining)
		n.pErr = roachpb.NewError(err)
		if n.pErr != nil {
			return false
		}
		n.colID = ColumnID(v)
		if qval, ok := n.qvals[n.colID]; ok && qval.datum == nil {
			value, ok = n.unmarshalValue(kv)
			if !ok {
				return false
			}
			qval.datum = value
			if log.V(2) {
				log.Infof("Scan %s -> %v", kv.Key, value)
			}
		} else {
			// No need to unmarshal the column value. Either the column was part of
			// the index key or it isn't needed by any of the render or filter
			// expressions.
			if log.V(2) {
				log.Infof("Scan %s -> [%d] (skipped)", kv.Key, n.colID)
			}
		}
	} else {
		if n.implicitVals != nil {
			implicitDirs := make([]encoding.Direction, 0, len(n.index.ImplicitColumnIDs))
			for range n.index.ImplicitColumnIDs {
				implicitDirs = append(implicitDirs, encoding.Ascending)
			}
			if _, n.pErr = decodeKeyVals(
				n.implicitValTypes, n.implicitVals, implicitDirs, kv.ValueBytes()); n.pErr != nil {
				return false
			}
			for i, id := range n.index.ImplicitColumnIDs {
				if qval, ok := n.qvals[id]; ok {
					qval.datum = n.implicitVals[i]
				}
			}
		}

		if log.V(2) {
			if n.implicitVals != nil {
				log.Infof("Scan %s -> %s", kv.Key, prettyDatums(n.implicitVals))
			} else {
				log.Infof("Scan %s", kv.Key)
			}
		}
	}

	if n.explain == explainDebug {
		if value == nil {
			if n.colID > 0 {
				var ok bool
				value, ok = n.unmarshalValue(kv)
				if !ok {
					return false
				}
			} else {
				value = parser.DNull
			}
		}
		n.explainValue = value
	}
	return true
}

// ProcessKV processes the given key/value, setting values in the row
// accordingly. If debugStrings is true, returns pretty printed key and value
// information in prettyKey/prettyValue (otherwise they are empty strings).
func (rf *RowFetcher) ProcessKV(kv client.KeyValue, debugStrings bool) (
	prettyKey string, prettyValue string, err error,
) {
	remaining, err := rf.ReadIndexKey(kv.Key)
	if err != nil {
		return "", "", err
	}

	if debugStrings {
		prettyKey = fmt.Sprintf("/%s/%s%s", rf.desc.Name, rf.index.Name, prettyDatums(rf.keyVals))
	}

	if rf.indexKey == nil {
		// This is the first key for the row.
		rf.indexKey = []byte(kv.Key[:len(kv.Key)-len(remaining)])

		// Reset the row to nil; it will get filled in with the column
		// values as we decode the key-value pairs for the row.
		for i := range rf.row {
			rf.row[i] = nil
		}

		// Fill in the column values that are part of the index key.
		for i, v := range rf.keyVals {
			rf.row[rf.indexColIdx[i]] = v
		}
	}

	if !rf.isSecondaryIndex && len(remaining) > 0 {
		_, familyID, err := encoding.DecodeUvarintAscending(remaining)
		if err != nil {
			return "", "", err
		}

		family, err := rf.desc.FindFamilyByID(FamilyID(familyID))
		if err != nil {
			return "", "", err
		}

		switch kv.Value.GetTag() {
		case roachpb.ValueType_TUPLE:
			prettyKey, prettyValue, err = rf.processValueTuple(family, kv, debugStrings, prettyKey)
		default:
			prettyKey, prettyValue, err = rf.processValueSingle(family, kv, debugStrings, prettyKey)
		}
		if err != nil {
			return "", "", err
		}
	} else {
		if rf.implicitVals != nil {
			// This is a unique index; decode the implicit column values from
			// the value.
			_, err := DecodeKeyVals(&rf.alloc, rf.implicitValTypes, rf.implicitVals, nil,
				kv.ValueBytes())
			if err != nil {
				return "", "", err
			}
			for i, id := range rf.index.ImplicitColumnIDs {
				if idx, ok := rf.colIdxMap[id]; ok && rf.valNeededForCol[idx] {
					rf.row[idx] = rf.implicitVals[i]
				}
			}
			if debugStrings {
				prettyValue = prettyDatums(rf.implicitVals)
			}
		}

		if log.V(2) {
			if rf.implicitVals != nil {
				log.Infof("Scan %s -> %s", kv.Key, prettyDatums(rf.implicitVals))
			} else {
				log.Infof("Scan %s", kv.Key)
			}
		}
	}

	if debugStrings && prettyValue == "" {
		prettyValue = parser.DNull.String()
	}

	return prettyKey, prettyValue, nil
}

// ProcessKV processes the given key/value, setting values in the row
// accordingly. If debugStrings is true, returns pretty printed key and value
// information in prettyKey/prettyValue (otherwise they are empty strings).
func (rf *RowFetcher) ProcessKV(kv client.KeyValue, debugStrings bool) (
	prettyKey string, prettyValue string, err error,
) {
	remaining, err := rf.ReadIndexKey(kv.Key)
	if err != nil {
		return "", "", err
	}

	if debugStrings {
		prettyKey = fmt.Sprintf("/%s/%s%s", rf.desc.Name, rf.index.Name, prettyDatums(rf.keyVals))
	}

	if rf.indexKey == nil {
		// This is the first key for the row.
		rf.indexKey = []byte(kv.Key[:len(kv.Key)-len(remaining)])

		// Reset the row to nil; it will get filled in with the column
		// values as we decode the key-value pairs for the row.
		for i := range rf.row {
			rf.row[i] = nil
		}

		// Fill in the column values that are part of the index key.
		for i, v := range rf.keyVals {
			rf.row[rf.indexColIdx[i]] = v
		}
	}

	if !rf.isSecondaryIndex && len(remaining) > 0 {
		_, colID, err := encoding.DecodeUvarintAscending(remaining)
		if err != nil {
			return "", "", err
		}

		idx, ok := rf.colIdxMap[ColumnID(colID)]
		if ok && (debugStrings || rf.valNeededForCol[idx]) {
			if debugStrings {
				prettyKey = fmt.Sprintf("%s/%s", prettyKey, rf.desc.Columns[idx].Name)
			}
			kind := rf.desc.Columns[idx].Type.Kind
			value, err := UnmarshalColumnValue(&rf.alloc, kind, kv.Value)
			if err != nil {
				return "", "", err
			}
			prettyValue = value.String()
			if rf.row[idx] != nil {
				panic(fmt.Sprintf("duplicate value for column %d", idx))
			}
			rf.row[idx] = value
			if log.V(3) {
				log.Infof("Scan %s -> %v", kv.Key, value)
			}
		} else {
			// No need to unmarshal the column value. Either the column was part of
			// the index key or it isn't needed.
			if log.V(3) {
				log.Infof("Scan %s -> [%d] (skipped)", kv.Key, colID)
			}
		}
	} else {
		if rf.implicitVals != nil {
			// This is a unique index; decode the implicit column values from
			// the value.
			_, err := DecodeKeyVals(&rf.alloc, rf.implicitValTypes, rf.implicitVals, nil,
				kv.ValueBytes())
			if err != nil {
				return "", "", err
			}
			for i, id := range rf.index.ImplicitColumnIDs {
				if idx, ok := rf.colIdxMap[id]; ok && rf.valNeededForCol[idx] {
					rf.row[idx] = rf.implicitVals[i]
				}
			}
			if debugStrings {
				prettyValue = prettyDatums(rf.implicitVals)
			}
		}

		if log.V(2) {
			if rf.implicitVals != nil {
				log.Infof("Scan %s -> %s", kv.Key, prettyDatums(rf.implicitVals))
			} else {
				log.Infof("Scan %s", kv.Key)
			}
		}
	}

	if debugStrings && prettyValue == "" {
		prettyValue = parser.DNull.String()
	}

	return prettyKey, prettyValue, nil
}

// DecodeTablePrefix validates that the given key has a table prefix, returning
// the remainder of the key (with the prefix removed) and the decoded descriptor
// ID of the table.
func DecodeTablePrefix(key roachpb.Key) ([]byte, uint64, error) {
	if encoding.PeekType(key) != encoding.Int {
		return key, 0, errors.Errorf("invalid key prefix: %q", key)
	}
	return encoding.DecodeUvarintAscending(key)
}