func TestPseudoRand(t *testing.T) {
	numbers := make(map[int]bool)
	// Make two random number generators and pull two numbers from each.
	rand1, _ := randutil.NewPseudoRand()
	rand2, _ := randutil.NewPseudoRand()
	numbers[rand1.Int()] = true
	numbers[rand1.Int()] = true
	numbers[rand2.Int()] = true
	numbers[rand2.Int()] = true
	// All four numbers should be distinct; no seed state is shared.
	if len(numbers) != 4 {
		t.Errorf("expected 4 unique numbers; got %d", len(numbers))
	}
}

func TestNonsortingEncodeDecimalRand(t *testing.T) {
	rng, _ := randutil.NewPseudoRand()
	const randomTrials = 200000
	for i := 0; i < randomTrials; i++ {
		var tmp, appendTo []byte
		// Test with and without appending.
		if rng.Intn(2) == 1 {
			appendTo = randBuf(rng, 30)
			appendTo = appendTo[:rng.Intn(len(appendTo)+1)]
		}
		// Test with and without tmp buffer.
		if rng.Intn(2) == 1 {
			tmp = randBuf(rng, 100)
		}
		cur := randDecimal(rng, -20, 20)

		enc := EncodeNonsortingDecimal(appendTo, cur)
		enc = enc[len(appendTo):]
		res, err := DecodeNonsortingDecimal(enc, tmp)
		if err != nil {
			t.Fatal(err)
		}

		// Make sure we decode the same value we encoded.
		if cur.Cmp(res) != 0 {
			t.Fatalf("unexpected mismatch for %v, got %v", cur, res)
		}

		// Make sure we would have overestimated the value.
		if est := UpperBoundNonsortingDecimalSize(cur); est < len(enc) {
			t.Fatalf("expected estimate of %d for %v to be greater than or equal to the encoded length, found [% x]", est, cur, enc)
		}
	}
}

func TestEncodeDecimalRand(t *testing.T) {
	rng, _ := randutil.NewPseudoRand()
	// Test both directions.
	for _, dir := range []Direction{Ascending, Descending} {
		var prev *inf.Dec
		var prevEnc []byte
		const randomTrials = 100000
		for i := 0; i < randomTrials; i++ {
			cur := randDecimal(rng, -20, 20)
			var tmp, appendTo []byte
			// Test with and without appending.
			if rng.Intn(2) == 1 {
				appendTo = randBuf(rng, 30)
				appendTo = appendTo[:rng.Intn(len(appendTo)+1)]
			}
			// Test with and without tmp buffer.
			if rng.Intn(2) == 1 {
				tmp = randBuf(rng, 100)
			}
			var enc []byte
			var res *inf.Dec
			var err error
			if dir == Ascending {
				enc = EncodeDecimalAscending(appendTo, cur)
				enc = enc[len(appendTo):]
				_, res, err = DecodeDecimalAscending(enc, tmp)
			} else {
				enc = EncodeDecimalDescending(appendTo, cur)
				enc = enc[len(appendTo):]
				_, res, err = DecodeDecimalDescending(enc, tmp)
			}
			if err != nil {
				t.Fatal(err)
			}

			testPeekLength(t, enc)

			// Make sure we decode the same value we encoded.
			if cur.Cmp(res) != 0 {
				t.Fatalf("unexpected mismatch for %v, got %v", cur, res)
			}

			// Make sure lexicographical sorting is consistent.
			if prev != nil {
				bytesCmp := bytes.Compare(prevEnc, enc)
				cmpType := "same"
				if dir == Descending {
					bytesCmp *= -1
					cmpType = "inverse"
				}
				if decCmp := prev.Cmp(cur); decCmp != bytesCmp {
					t.Fatalf("expected [% x] to compare to [% x] the %s way that %v compares to %v",
						prevEnc, enc, cmpType, prev, cur)
				}
			}
			prev = cur
			prevEnc = enc
		}
	}
}

// TestTransactionObservedTimestamp verifies that txn.{Get,Update}ObservedTimestamp work as
// advertised.
func TestTransactionObservedTimestamp(t *testing.T) {
	var txn Transaction
	rng, seed := randutil.NewPseudoRand()
	t.Logf("running with seed %d", seed)
	ids := append([]int{109, 104, 102, 108, 1000}, rand.Perm(100)...)
	timestamps := make(map[NodeID]hlc.Timestamp, len(ids))
	for i := 0; i < len(ids); i++ {
		timestamps[NodeID(i)] = hlc.ZeroTimestamp.Add(rng.Int63(), 0)
	}
	for i, n := range ids {
		nodeID := NodeID(n)
		if ts, ok := txn.GetObservedTimestamp(nodeID); ok {
			t.Fatalf("%d: false positive hit %s in %v", nodeID, ts, ids[:i+1])
		}
		txn.UpdateObservedTimestamp(nodeID, timestamps[nodeID])
		txn.UpdateObservedTimestamp(nodeID, hlc.MaxTimestamp) // should be noop
		if exp, act := i+1, len(txn.ObservedTimestamps); act != exp {
			t.Fatalf("%d: expected %d entries, got %d: %v", nodeID, exp, act, txn.ObservedTimestamps)
		}
	}
	for _, m := range ids {
		checkID := NodeID(m)
		exp := timestamps[checkID]
		if act, _ := txn.GetObservedTimestamp(checkID); !act.Equal(exp) {
			t.Fatalf("%d: expected %s, got %s", checkID, exp, act)
		}
	}

	var emptyTxn Transaction
	ts := hlc.ZeroTimestamp.Add(1, 2)
	emptyTxn.UpdateObservedTimestamp(NodeID(1), ts)
	if actTS, _ := emptyTxn.GetObservedTimestamp(NodeID(1)); !actTS.Equal(ts) {
		t.Fatalf("unexpected: %s (wanted %s)", actTS, ts)
	}
}

func TestEncDatumNull(t *testing.T) {
	defer leaktest.AfterTest(t)()

	// Verify DNull is null.
	n := DatumToEncDatum(ColumnType{Kind: ColumnType_INT}, parser.DNull)
	if !n.IsNull() {
		t.Error("DNull not null")
	}

	var alloc DatumAlloc
	rng, _ := randutil.NewPseudoRand()

	// Generate random EncDatums (some of which are null), and verify that a datum
	// created from its encoding has the same IsNull() value.
	for cases := 0; cases < 100; cases++ {
		a := RandEncDatum(rng)

		for enc := range DatumEncoding_name {
			encoded, err := a.Encode(&alloc, DatumEncoding(enc), nil)
			if err != nil {
				t.Fatal(err)
			}
			b := EncDatumFromEncoded(ColumnType{Kind: ColumnType_INT}, DatumEncoding(enc), encoded)
			if a.IsNull() != b.IsNull() {
				t.Errorf("before: %s (null=%t) after: %s (null=%t)",
					a.String(), a.IsNull(), b.String(), b.IsNull())
			}
		}
	}

}

// TestStreamEncodeDecode generates random streams of EncDatums and passes them
// through a StreamEncoder and a StreamDecoder
func TestStreamEncodeDecode(t *testing.T) {
	defer leaktest.AfterTest(t)()
	rng, _ := randutil.NewPseudoRand()
	for test := 0; test < 100; test++ {
		rowLen := 1 + rng.Intn(20)
		info := make([]DatumInfo, rowLen)
		for i := range info {
			info[i].Type = sqlbase.RandColumnType(rng)
			info[i].Encoding = sqlbase.RandDatumEncoding(rng)
		}
		numRows := rng.Intn(100)
		rows := make(sqlbase.EncDatumRows, numRows)
		for i := range rows {
			rows[i] = make(sqlbase.EncDatumRow, rowLen)
			for j := range rows[i] {
				rows[i][j] = sqlbase.DatumToEncDatum(info[j].Type,
					sqlbase.RandDatum(rng, info[j].Type, true))
			}
		}
		var trailerErr error
		if rng.Intn(10) == 0 {
			trailerErr = fmt.Errorf("test error %d", rng.Intn(100))
		}
		testRowStream(t, rng, rows, trailerErr)
	}
}

func runMVCCConditionalPut(emk engineMaker, valueSize int, createFirst bool, b *testing.B) {
	rng, _ := randutil.NewPseudoRand()
	value := roachpb.MakeValueFromBytes(randutil.RandBytes(rng, valueSize))
	keyBuf := append(make([]byte, 0, 64), []byte("key-")...)

	eng := emk(b, fmt.Sprintf("cput_%d", valueSize))
	defer eng.Close()

	b.SetBytes(int64(valueSize))
	var expected *roachpb.Value
	if createFirst {
		for i := 0; i < b.N; i++ {
			key := roachpb.Key(encoding.EncodeUvarintAscending(keyBuf[:4], uint64(i)))
			ts := makeTS(timeutil.Now().UnixNano(), 0)
			if err := MVCCPut(context.Background(), eng, nil, key, ts, value, nil); err != nil {
				b.Fatalf("failed put: %s", err)
			}
		}
		expected = &value
	}

	b.ResetTimer()

	for i := 0; i < b.N; i++ {
		key := roachpb.Key(encoding.EncodeUvarintAscending(keyBuf[:4], uint64(i)))
		ts := makeTS(timeutil.Now().UnixNano(), 0)
		if err := MVCCConditionalPut(context.Background(), eng, nil, key, ts, value, expected, nil); err != nil {
			b.Fatalf("failed put: %s", err)
		}
	}

	b.StopTimer()
}

func makeEncodedVals(minExp, maxExp int) [][]byte {
	rng, _ := randutil.NewPseudoRand()
	vals := make([][]byte, 10000)
	for i := range vals {
		vals[i] = EncodeDecimalAscending(nil, randDecimal(rng, minExp, maxExp))
	}
	return vals
}

// makeDecimalVals creates decimal values with exponents in
// the range [minExp, maxExp].
func makeDecimalVals(minExp, maxExp int) []*inf.Dec {
	rng, _ := randutil.NewPseudoRand()
	vals := make([]*inf.Dec, 10000)
	for i := range vals {
		vals[i] = randDecimal(rng, minExp, maxExp)
	}
	return vals
}

func makeIntTestDatum(count int) []Datum {
	rng, _ := randutil.NewPseudoRand()

	vals := make([]Datum, count)
	for i := range vals {
		vals[i] = NewDInt(DInt(rng.Int63()))
	}
	return vals
}

func makeFloatTestDatum(count int) []Datum {
	rng, _ := randutil.NewPseudoRand()

	vals := make([]Datum, count)
	for i := range vals {
		vals[i] = NewDFloat(DFloat(rng.Float64()))
	}
	return vals
}

func TestRandBytes(t *testing.T) {
	rand, _ := randutil.NewPseudoRand()
	for i := 0; i < 100; i++ {
		x := randutil.RandBytes(rand, i)
		if len(x) != i {
			t.Errorf("got array with unexpected length: %d (expected %d)", len(x), i)
		}
	}
}

func makeBoolTestDatum(count int) []Datum {
	rng, _ := randutil.NewPseudoRand()

	vals := make([]Datum, count)
	for i := range vals {
		vals[i] = MakeDBool(DBool(rng.Int31n(2) == 0))
	}
	return vals
}

func TestRandIntInRange(t *testing.T) {
	rand, _ := randutil.NewPseudoRand()
	for i := 0; i < 100; i++ {
		x := randutil.RandIntInRange(rand, 20, 40)
		if x < 20 || x >= 40 {
			t.Errorf("got result out of range: %d", x)
		}
	}
}

func init() {
	r, _ := randutil.NewPseudoRand()
	// 127.255.255.255 is special (broadcast), so choose values less
	// than 255.
	a := r.Intn(255)
	b := r.Intn(255)
	c := r.Intn(255)
	IsolatedTestAddr = NewUnresolvedAddr("tcp", fmt.Sprintf("127.%d.%d.%d:0", a, b, c))
}

func makeDecimalTestDatum(count int) []Datum {
	rng, _ := randutil.NewPseudoRand()

	vals := make([]Datum, count)
	for i := range vals {
		dd := &DDecimal{}
		decimal.SetFromFloat(&dd.Dec, rng.Float64())
		vals[i] = dd
	}
	return vals
}

func TestEncDatumCompare(t *testing.T) {
	defer leaktest.AfterTest(t)()

	a := &DatumAlloc{}
	rng, _ := randutil.NewPseudoRand()

	for kind := range ColumnType_Kind_name {
		kind := ColumnType_Kind(kind)
		// TODO(cuongdo,eisen): we don't support persistence for arrays or collated
		// strings yet
		if kind == ColumnType_COLLATEDSTRING || kind == ColumnType_INT_ARRAY {
			continue
		}
		typ := ColumnType{Kind: kind}

		// Generate two datums d1 < d2
		var d1, d2 parser.Datum
		for {
			d1 = RandDatum(rng, typ, false)
			d2 = RandDatum(rng, typ, false)
			if cmp := d1.Compare(d2); cmp < 0 {
				break
			}
		}
		v1 := DatumToEncDatum(typ, d1)
		v2 := DatumToEncDatum(typ, d2)

		if val, err := v1.Compare(a, &v2); err != nil {
			t.Fatal(err)
		} else if val != -1 {
			t.Errorf("compare(1, 2) = %d", val)
		}

		asc := DatumEncoding_ASCENDING_KEY
		desc := DatumEncoding_DESCENDING_KEY
		noncmp := DatumEncoding_VALUE

		checkEncDatumCmp(t, a, &v1, &v2, asc, asc, -1, false)
		checkEncDatumCmp(t, a, &v2, &v1, asc, asc, +1, false)
		checkEncDatumCmp(t, a, &v1, &v1, asc, asc, 0, false)
		checkEncDatumCmp(t, a, &v2, &v2, asc, asc, 0, false)

		checkEncDatumCmp(t, a, &v1, &v2, desc, desc, -1, false)
		checkEncDatumCmp(t, a, &v2, &v1, desc, desc, +1, false)
		checkEncDatumCmp(t, a, &v1, &v1, desc, desc, 0, false)
		checkEncDatumCmp(t, a, &v2, &v2, desc, desc, 0, false)

		checkEncDatumCmp(t, a, &v1, &v2, noncmp, noncmp, -1, true)
		checkEncDatumCmp(t, a, &v2, &v1, desc, noncmp, +1, true)
		checkEncDatumCmp(t, a, &v1, &v1, asc, desc, 0, true)
		checkEncDatumCmp(t, a, &v2, &v2, desc, asc, 0, true)
	}
}

func testPutInner(ctx context.Context, t *testing.T, c cluster.Cluster, cfg cluster.TestConfig) {
	db, err := c.NewClient(ctx, 0)
	if err != nil {
		t.Fatal(err)
	}

	errs := make(chan error, c.NumNodes())
	start := timeutil.Now()
	deadline := start.Add(cfg.Duration)
	var count int64
	for i := 0; i < c.NumNodes(); i++ {
		go func() {
			r, _ := randutil.NewPseudoRand()
			value := randutil.RandBytes(r, 8192)

			for timeutil.Now().Before(deadline) {
				k := atomic.AddInt64(&count, 1)
				v := value[:r.Intn(len(value))]
				if err := db.Put(ctx, fmt.Sprintf("%08d", k), v); err != nil {
					errs <- err
					return
				}
			}
			errs <- nil
		}()
	}

	for i := 0; i < c.NumNodes(); {
		baseCount := atomic.LoadInt64(&count)
		select {
		case <-stopper.ShouldStop():
			t.Fatalf("interrupted")
		case err := <-errs:
			if err != nil {
				t.Fatal(err)
			}
			i++
		case <-time.After(1 * time.Second):
			// Periodically print out progress so that we know the test is still
			// running.
			loadedCount := atomic.LoadInt64(&count)
			log.Infof(ctx, "%d (%d/s)", loadedCount, loadedCount-baseCount)
			c.Assert(ctx, t)
			if err := cluster.Consistent(ctx, c, 0); err != nil {
				t.Fatal(err)
			}
		}
	}

	elapsed := timeutil.Since(start)
	log.Infof(ctx, "%d %.1f/sec", count, float64(count)/elapsed.Seconds())
}

// makeSmallIntTestDatum creates integers that are sufficiently
// smaller than 2^64-1 that they can be added to each other for a
// significant part of the test without overflow. This is meant to
// test the implementation of aggregates that can use an int64 to
// optimize computations small decimal values.
func makeSmallIntTestDatum(count int) []Datum {
	rng, _ := randutil.NewPseudoRand()

	vals := make([]Datum, count)
	for i := range vals {
		sign := int32(1)
		if 0 == rng.Int31()&1 {
			sign = -1
		}
		vals[i] = NewDInt(DInt(rng.Int31() * sign))
	}
	return vals
}

func BenchmarkDecodeFloat(b *testing.B) {
	rng, _ := randutil.NewPseudoRand()

	vals := make([][]byte, 10000)
	for i := range vals {
		vals[i] = EncodeFloatAscending(nil, rng.Float64())
	}

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		_, _, _ = DecodeFloatAscending(vals[i%len(vals)])
	}
}