// TestGossipRestart verifies that the gossip network can be
// re-bootstrapped after a time when all nodes were down
// simultaneously.
func TestGossipRestart(t *testing.T) {
	// This already replicates the first range (in the local setup).
	// The replication of the first range is important: as long as the
	// first range only exists on one node, that node can trivially
	// acquire the leader lease. Once the range is replicated, however,
	// nodes must be able to discover each other over gossip before the
	// lease can be acquired.
	c := StartCluster(t)
	defer c.AssertAndStop(t)
	num := c.NumNodes()

	log.Infof("waiting for initial gossip connections")
	checkGossip(t, c, 20*time.Second, hasPeers(num))
	checkGossip(t, c, time.Second, hasClusterID)
	checkGossip(t, c, time.Second, hasSentinel)

	log.Infof("killing all nodes")
	for i := 0; i < num; i++ {
		if err := c.Kill(i); err != nil {
			t.Fatal(err)
		}
	}

	log.Infof("restarting all nodes")
	for i := 0; i < num; i++ {
		if err := c.Restart(i); err != nil {
			t.Fatal(err)
		}
	}

	log.Infof("waiting for gossip to be connected")
	checkGossip(t, c, 20*time.Second, hasPeers(num))
	checkGossip(t, c, time.Second, hasClusterID)
	checkGossip(t, c, time.Second, hasSentinel)

	for i := 0; i < num; i++ {
		db, dbStopper := c.MakeClient(t, i)
		if i == 0 {
			if err := db.Del("count"); err != nil {
				t.Fatal(err)
			}
		}
		var kv client.KeyValue
		if err := db.Txn(func(txn *client.Txn) error {
			var err error
			kv, err = txn.Inc("count", 1)
			return err
		}); err != nil {
			t.Fatal(err)
		} else if v := kv.ValueInt(); v != int64(i+1) {
			t.Fatalf("unexpected value %d for write #%d (expected %d)", v, i, i+1)
		}
		dbStopper.Stop()
	}
}

func (ia *idAllocator) start() {
	ia.stopper.RunWorker(func() {
		defer close(ia.ids)

		for {
			var newValue int64
			for newValue <= int64(ia.minID) {
				var (
					err error
					res client.KeyValue
				)
				for r := retry.Start(idAllocationRetryOpts); r.Next(); {
					if ia.stopper.StartTask() {
						idKey := ia.idKey.Load().(proto.Key)
						res, err = ia.db.Inc(idKey, int64(ia.blockSize))
						ia.stopper.FinishTask()

						if err == nil {
							newValue = res.ValueInt()
							break
						}

						log.Warningf("unable to allocate %d ids from %s: %s", ia.blockSize, idKey, err)
					} else {
						return
					}
				}
				if err != nil {
					panic(fmt.Sprintf("unexpectedly exited id allocation retry loop: %s", err))
				}
			}

			end := newValue + 1
			start := end - int64(ia.blockSize)

			if start < int64(ia.minID) {
				start = int64(ia.minID)
			}

			// Add all new ids to the channel for consumption.
			for i := start; i < end; i++ {
				select {
				case ia.ids <- uint32(i):
				case <-ia.stopper.ShouldStop():
					return
				}
			}
		}
	})
}

func unmarshalValue(col structured.ColumnDescriptor, kv client.KeyValue) driver.Value {
	if !kv.Exists() {
		return nil
	}
	switch col.Type.Kind {
	case structured.ColumnType_BIT, structured.ColumnType_INT:
		return kv.ValueInt()
	case structured.ColumnType_FLOAT:
		return math.Float64frombits(uint64(kv.ValueInt()))
	case structured.ColumnType_CHAR, structured.ColumnType_BINARY,
		structured.ColumnType_TEXT, structured.ColumnType_BLOB:
		// TODO(pmattis): The conversion to string isn't strictly necessary, but
		// makes log messages more readable right now.
		return string(kv.ValueBytes())
	}
	return kv.ValueBytes()
}

func (n *scanNode) unmarshalValue(kv client.KeyValue) (parser.Datum, bool) {
	kind, ok := n.colKind[n.colID]
	if !ok {
		n.err = fmt.Errorf("column-id \"%d\" does not exist", n.colID)
		return nil, false
	}
	if kv.Exists() {
		switch kind {
		case ColumnType_INT:
			return parser.DInt(kv.ValueInt()), true
		case ColumnType_BOOL:
			return parser.DBool(kv.ValueInt() != 0), true
		case ColumnType_FLOAT:
			return parser.DFloat(math.Float64frombits(uint64(kv.ValueInt()))), true
		case ColumnType_STRING, ColumnType_BYTES:
			return parser.DString(kv.ValueBytes()), true
		}
	}
	return parser.DNull, true
}

func unmarshalValue(col structured.ColumnDescriptor, kv client.KeyValue) parser.Datum {
	if kv.Exists() {
		switch col.Type.Kind {
		case structured.ColumnType_BIT, structured.ColumnType_INT:
			return parser.DInt(kv.ValueInt())
		case structured.ColumnType_FLOAT:
			return parser.DFloat(math.Float64frombits(uint64(kv.ValueInt())))
		case structured.ColumnType_CHAR, structured.ColumnType_TEXT,
			structured.ColumnType_BLOB:
			return parser.DString(kv.ValueBytes())
		}
	}
	return parser.DNull
}

func unmarshalValue(col structured.ColumnDescriptor, kv client.KeyValue) sqlwire.Datum {
	var d sqlwire.Datum
	if !kv.Exists() {
		return d
	}
	switch col.Type.Kind {
	case structured.ColumnType_BIT, structured.ColumnType_INT:
		tmp := kv.ValueInt()
		d.IntVal = &tmp
	case structured.ColumnType_FLOAT:
		tmp := math.Float64frombits(uint64(kv.ValueInt()))
		d.FloatVal = &tmp
	case structured.ColumnType_CHAR, structured.ColumnType_TEXT,
		structured.ColumnType_BLOB:
		// TODO(pmattis): The conversion to string isn't strictly necessary, but
		// makes log messages more readable right now.
		tmp := string(kv.ValueBytes())
		d.StringVal = &tmp
	}
	return d
}

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 expresssions 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.err = decodeIndexKey(n.desc, *n.index, n.valTypes, n.vals, kv.Key)
	if n.err != 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
		_, v, n.err = encoding.DecodeUvarint(remaining)
		if n.err != 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", prettyKey(kv.Key, 0), 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)", prettyKey(kv.Key, 0), n.colID)
			}
		}
	} else {
		if n.implicitVals != nil {
			if _, n.err = decodeKeyVals(n.implicitValTypes, n.implicitVals, kv.ValueBytes()); n.err != 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", prettyKey(kv.Key, 0), prettyKeyVals(n.implicitVals))
			} else {
				log.Infof("Scan %s", prettyKey(kv.Key, 0))
			}
		}
	}

	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
}

func (n *scanNode) unmarshalValue(kv client.KeyValue) (parser.Datum, bool) {
	kind, ok := n.colKind[n.colID]
	if !ok {
		n.err = fmt.Errorf("column-id \"%d\" does not exist", n.colID)
		return nil, false
	}
	if kv.Exists() {
		switch kind {
		case ColumnType_INT:
			return parser.DInt(kv.ValueInt()), true
		case ColumnType_BOOL:
			return parser.DBool(kv.ValueInt() != 0), true
		case ColumnType_FLOAT:
			return parser.DFloat(math.Float64frombits(uint64(kv.ValueInt()))), true
		case ColumnType_STRING:
			return parser.DString(kv.ValueBytes()), true
		case ColumnType_BYTES:
			return parser.DBytes(kv.ValueBytes()), true
		case ColumnType_DATE:
			var t time.Time
			if err := t.UnmarshalBinary(kv.ValueBytes()); err != nil {
				return nil, false
			}
			return parser.DDate{Time: t}, true
		case ColumnType_TIMESTAMP:
			var t time.Time
			if err := t.UnmarshalBinary(kv.ValueBytes()); err != nil {
				return nil, false
			}
			return parser.DTimestamp{Time: t}, true
		case ColumnType_INTERVAL:
			return parser.DInterval{Duration: time.Duration(kv.ValueInt())}, true
		}
	}
	return parser.DNull, 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 {
		_, 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
}

// 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
}

func testGossipRestartInner(t *testing.T, c cluster.Cluster, cfg cluster.TestConfig) {
	// This already replicates the first range (in the local setup).
	// The replication of the first range is important: as long as the
	// first range only exists on one node, that node can trivially
	// acquire the leader lease. Once the range is replicated, however,
	// nodes must be able to discover each other over gossip before the
	// lease can be acquired.
	num := c.NumNodes()

	deadline := timeutil.Now().Add(cfg.Duration)

	waitTime := longWaitTime
	if cfg.Duration < waitTime {
		waitTime = shortWaitTime
	}

	for timeutil.Now().Before(deadline) {
		log.Infof("waiting for initial gossip connections")
		checkGossip(t, c, waitTime, hasPeers(num))
		checkGossip(t, c, waitTime, hasClusterID)
		checkGossip(t, c, waitTime, hasSentinel)

		log.Infof("killing all nodes")
		for i := 0; i < num; i++ {
			if err := c.Kill(i); err != nil {
				t.Fatal(err)
			}
		}

		log.Infof("restarting all nodes")
		for i := 0; i < num; i++ {
			if err := c.Restart(i); err != nil {
				t.Fatal(err)
			}
		}

		log.Infof("waiting for gossip to be connected")
		checkGossip(t, c, waitTime, hasPeers(num))
		checkGossip(t, c, waitTime, hasClusterID)
		checkGossip(t, c, waitTime, hasSentinel)

		for i := 0; i < num; i++ {
			db, dbStopper := c.NewClient(t, i)
			if i == 0 {
				if err := db.Del("count"); err != nil {
					t.Fatal(err)
				}
			}
			var kv client.KeyValue
			if err := db.Txn(func(txn *client.Txn) error {
				var err error
				kv, err = txn.Inc("count", 1)
				return err
			}); err != nil {
				t.Fatal(err)
			} else if v := kv.ValueInt(); v != int64(i+1) {
				t.Fatalf("unexpected value %d for write #%d (expected %d)", v, i, i+1)
			}
			dbStopper.Stop()
		}
	}
}