// TestStoreRangeSplitWithMaxBytesUpdate tests a scenario where a new
// zone config that updates the max bytes is set and triggers a range
// split.
func TestStoreRangeSplitWithMaxBytesUpdate(t *testing.T) {
	defer leaktest.AfterTest(t)
	store, stopper := createTestStore(t)
	config.TestingSetupZoneConfigHook(stopper)
	defer stopper.Stop()

	origRng := store.LookupReplica(roachpb.RKeyMin, nil)

	// Set max bytes.
	maxBytes := int64(1 << 16)
	config.TestingSetZoneConfig(1000, &config.ZoneConfig{RangeMaxBytes: maxBytes})

	// Trigger gossip callback.
	if err := store.Gossip().AddInfoProto(gossip.KeySystemConfig, &config.SystemConfig{}, 0); err != nil {
		t.Fatal(err)
	}

	// Verify that the range is split and the new range has the correct max bytes.
	util.SucceedsWithin(t, time.Second, func() error {
		newRng := store.LookupReplica(keys.MakeTablePrefix(1000), nil)
		if newRng.Desc().RangeID == origRng.Desc().RangeID {
			return util.Errorf("expected new range created by split")
		}
		if newRng.GetMaxBytes() != maxBytes {
			return util.Errorf("expected %d max bytes for the new range, but got %d",
				maxBytes, newRng.GetMaxBytes())
		}
		return nil
	})
}

// TestStoreZoneUpdateAndRangeSplit verifies that modifying the zone
// configuration changes range max bytes and Range.maybeSplit() takes
// max bytes into account when deciding whether to enqueue a range for
// splitting. It further verifies that the range is in fact split on
// exceeding zone's RangeMaxBytes.
func TestStoreZoneUpdateAndRangeSplit(t *testing.T) {
	defer leaktest.AfterTest(t)
	t.Skip("#3762")
	store, stopper := createTestStore(t)
	config.TestingSetupZoneConfigHook(stopper)
	defer stopper.Stop()

	maxBytes := int64(1 << 16)
	// Set max bytes.
	descID := uint32(keys.MaxReservedDescID + 1)
	config.TestingSetZoneConfig(descID, &config.ZoneConfig{RangeMaxBytes: maxBytes})

	// Trigger gossip callback.
	if err := store.Gossip().AddInfoProto(gossip.KeySystemConfig, &config.SystemConfig{}, 0); err != nil {
		t.Fatal(err)
	}

	// Wait for the range to be split along table boundaries.
	originalRange := store.LookupReplica(roachpb.RKey(keys.UserTableDataMin), nil)
	var rng *storage.Replica
	if err := util.IsTrueWithin(func() bool {
		rng = store.LookupReplica(keys.MakeTablePrefix(descID), nil)
		return rng.RangeID != originalRange.RangeID
	}, splitTimeout); err != nil {
		t.Fatalf("failed to notice range max bytes update: %s", err)
	}

	// Check range's max bytes settings.
	if rng.GetMaxBytes() != maxBytes {
		t.Fatalf("range max bytes mismatch, got: %d, expected: %d", rng.GetMaxBytes(), maxBytes)
	}

	// Make sure the second range goes to the end.
	if !roachpb.RKeyMax.Equal(rng.Desc().EndKey) {
		t.Fatalf("second range has split: %+v", rng.Desc())
	}

	// Look in the range after prefix we're writing to.
	fillRange(store, rng.RangeID, keys.MakeTablePrefix(descID), maxBytes, t)

	// Verify that the range is in fact split (give it a few seconds for very
	// slow test machines).
	var newRng *storage.Replica
	util.SucceedsWithin(t, splitTimeout, func() error {
		newRng = store.LookupReplica(keys.MakeTablePrefix(descID+1), nil)
		if newRng.RangeID == rng.RangeID {
			return util.Errorf("range has not yet split")
		}
		return nil
	})

	// Make sure the new range goes to the end.
	if !roachpb.RKeyMax.Equal(newRng.Desc().EndKey) {
		t.Fatalf("second range has split: %+v", rng.Desc())
	}
}

// TestStoreZoneUpdateAndRangeSplit verifies that modifying the zone
// configuration changes range max bytes and Range.maybeSplit() takes
// max bytes into account when deciding whether to enqueue a range for
// splitting. It further verifies that the range is in fact split on
// exceeding zone's RangeMaxBytes.
func TestStoreZoneUpdateAndRangeSplit(t *testing.T) {
	defer leaktest.AfterTest(t)
	store, stopper := createTestStore(t)
	config.TestingSetupZoneConfigHook(stopper)
	defer stopper.Stop()

	maxBytes := int64(1 << 16)
	// Set max bytes.
	config.TestingSetZoneConfig(1000, &config.ZoneConfig{RangeMaxBytes: maxBytes})

	// Trigger gossip callback.
	if err := store.Gossip().AddInfoProto(gossip.KeySystemConfig, &config.SystemConfig{}, 0); err != nil {
		t.Fatal(err)
	}

	// Wait for the range to be split along table boundaries.
	originalRange := store.LookupReplica(roachpb.RKeyMin, nil)
	var rng *storage.Replica
	if err := util.IsTrueWithin(func() bool {
		rng = store.LookupReplica(keys.MakeTablePrefix(1000), nil)
		return rng.Desc().RangeID != originalRange.Desc().RangeID
	}, 50*time.Millisecond); err != nil {
		t.Fatalf("failed to notice range max bytes update: %s", err)
	}

	// Check range's max bytes settings.
	if rng.GetMaxBytes() != maxBytes {
		t.Fatalf("range max bytes mismatch, got: %d, expected: %d", rng.GetMaxBytes(), maxBytes)
	}

	// Make sure the second range goes to the end.
	if !roachpb.RKeyMax.Equal(rng.Desc().EndKey) {
		t.Fatalf("second range has split: %+v", rng.Desc())
	}

	// Look in the range after prefix we're writing to.
	fillRange(store, rng.Desc().RangeID, keys.MakeTablePrefix(1000), maxBytes, t)

	// Verify that the range is in fact split (give it a second for very slow test machines).
	var newRng *storage.Replica
	if err := util.IsTrueWithin(func() bool {
		newRng = store.LookupReplica(keys.MakeTablePrefix(2000), nil)
		return newRng.Desc().RangeID != rng.Desc().RangeID
	}, time.Second); err != nil {
		t.Errorf("expected range to split within 1s")
	}

	// Make sure the new range goes to the end.
	if !roachpb.RKeyMax.Equal(newRng.Desc().EndKey) {
		t.Fatalf("second range has split: %+v", rng.Desc())
	}
}

// TestStoreSetRangesMaxBytes creates a set of ranges via splitting
// and then sets the config zone to a custom max bytes value to
// verify the ranges' max bytes are updated appropriately.
func TestStoreSetRangesMaxBytes(t *testing.T) {
	defer leaktest.AfterTest(t)
	store, _, stopper := createTestStore(t)
	defer stopper.Stop()

	testData := []struct {
		rng         *Replica
		expMaxBytes int64
	}{
		{store.LookupReplica(roachpb.KeyMin, nil),
			config.DefaultZoneConfig.RangeMaxBytes},
		{splitTestRange(store, roachpb.KeyMin, keys.MakeTablePrefix(1000), t),
			1 << 20},
		{splitTestRange(store, keys.MakeTablePrefix(1000), keys.MakeTablePrefix(1001), t),
			config.DefaultZoneConfig.RangeMaxBytes},
		{splitTestRange(store, keys.MakeTablePrefix(1001), keys.MakeTablePrefix(1002), t),
			2 << 20},
	}

	// Set zone configs.
	config.TestingSetZoneConfig(1000, &config.ZoneConfig{RangeMaxBytes: 1 << 20})
	config.TestingSetZoneConfig(1002, &config.ZoneConfig{RangeMaxBytes: 2 << 20})

	// Despite faking the zone configs, we still need to have a gossip entry.
	if err := store.Gossip().AddInfoProto(gossip.KeySystemConfig, &config.SystemConfig{}, 0); err != nil {
		t.Fatal(err)
	}

	if err := util.IsTrueWithin(func() bool {
		for _, test := range testData {
			if test.rng.GetMaxBytes() != test.expMaxBytes {
				return false
			}
		}
		return true
	}, 500*time.Millisecond); err != nil {
		t.Errorf("range max bytes values did not change as expected: %s", err)
	}
}

// TestStoreZoneUpdateAndRangeSplit verifies that modifying the zone
// configuration changes range max bytes and Range.maybeSplit() takes
// max bytes into account when deciding whether to enqueue a range for
// splitting. It further verifies that the range is in fact split on
// exceeding zone's RangeMaxBytes.
func TestStoreZoneUpdateAndRangeSplit(t *testing.T) {
	defer leaktest.AfterTest(t)()
	store, stopper, _ := createTestStore(t)
	config.TestingSetupZoneConfigHook(stopper)
	defer stopper.Stop()

	maxBytes := int64(1 << 16)
	// Set max bytes.
	descID := uint32(keys.MaxReservedDescID + 1)
	config.TestingSetZoneConfig(descID, &config.ZoneConfig{RangeMaxBytes: maxBytes})

	// Trigger gossip callback.
	if err := store.Gossip().AddInfoProto(gossip.KeySystemConfig, &config.SystemConfig{}, 0); err != nil {
		t.Fatal(err)
	}

	tableBoundary := keys.MakeTablePrefix(descID)

	{
		var rng *storage.Replica

		// Wait for the range to be split along table boundaries.
		expectedRSpan := roachpb.RSpan{Key: roachpb.RKey(tableBoundary), EndKey: roachpb.RKeyMax}
		util.SucceedsSoon(t, func() error {
			rng = store.LookupReplica(tableBoundary, nil)
			if actualRSpan := rng.Desc().RSpan(); !actualRSpan.Equal(expectedRSpan) {
				return util.Errorf("expected range %s to span %s", rng, expectedRSpan)
			}
			return nil
		})

		// Check range's max bytes settings.
		if actualMaxBytes := rng.GetMaxBytes(); actualMaxBytes != maxBytes {
			t.Fatalf("range %s max bytes mismatch, got: %d, expected: %d", rng, actualMaxBytes, maxBytes)
		}

		// Look in the range after prefix we're writing to.
		fillRange(store, rng.RangeID, tableBoundary, maxBytes, t)
	}

	// Verify that the range is in fact split.
	util.SucceedsSoon(t, func() error {
		rng := store.LookupReplica(keys.MakeTablePrefix(descID+1), nil)
		rngDesc := rng.Desc()
		rngStart, rngEnd := rngDesc.StartKey, rngDesc.EndKey
		if rngStart.Equal(tableBoundary) || !rngEnd.Equal(roachpb.RKeyMax) {
			return util.Errorf("range %s has not yet split", rng)
		}
		return nil
	})
}

// TestAcceptsUnsplitRanges verifies that ranges that need to split are properly
// rejected when the queue has 'acceptsUnsplitRanges = false'.
func TestAcceptsUnsplitRanges(t *testing.T) {
	defer leaktest.AfterTest(t)
	g, stopper := gossipForTest(t)
	defer stopper.Stop()

	// This range can never be split due to zone configs boundaries.
	neverSplits := &Replica{}
	if err := neverSplits.setDesc(&roachpb.RangeDescriptor{
		RangeID:  1,
		StartKey: roachpb.RKeyMin,
		EndKey:   keys.Addr(keys.UserTableDataMin),
	}); err != nil {
		t.Fatal(err)
	}

	// This range will need to be split after user db/table entries are created.
	willSplit := &Replica{}
	if err := willSplit.setDesc(&roachpb.RangeDescriptor{
		RangeID:  2,
		StartKey: keys.Addr(keys.UserTableDataMin),
		EndKey:   roachpb.RKeyMax,
	}); err != nil {
		t.Fatal(err)
	}

	var queued int32
	testQueue := &testQueueImpl{
		shouldQueueFn: func(now roachpb.Timestamp, r *Replica) (shouldQueue bool, priority float64) {
			// Always queue ranges if they make it past the base queue's logic.
			atomic.AddInt32(&queued, 1)
			return true, float64(r.Desc().RangeID)
		},
		acceptUnsplit: false,
	}

	bq := makeBaseQueue("test", testQueue, g, 2)
	mc := hlc.NewManualClock(0)
	clock := hlc.NewClock(mc.UnixNano)
	bq.Start(clock, stopper)

	// Check our config.
	sysCfg := g.GetSystemConfig()
	if sysCfg == nil {
		t.Fatal("nil config")
	}
	if sysCfg.NeedsSplit(neverSplits.Desc().StartKey, neverSplits.Desc().EndKey) {
		t.Fatal("System config says range needs to be split")
	}
	if sysCfg.NeedsSplit(willSplit.Desc().StartKey, willSplit.Desc().EndKey) {
		t.Fatal("System config says range needs to be split")
	}

	// There are no user db/table entries, everything should be added and
	// processed as usual.
	bq.MaybeAdd(neverSplits, roachpb.ZeroTimestamp)
	bq.MaybeAdd(willSplit, roachpb.ZeroTimestamp)

	if err := util.IsTrueWithin(func() bool {
		return atomic.LoadInt32(&testQueue.processed) == 2
	}, 250*time.Millisecond); err != nil {
		t.Error(err)
	}

	if pc := atomic.LoadInt32(&queued); pc != 2 {
		t.Errorf("expected queued count of 2; got %d", pc)
	}

	// Now add a user object, it will trigger a split.
	// The range willSplit starts at the beginning of the user data range,
	// which means keys.MaxReservedDescID+1.
	config.TestingSetZoneConfig(keys.MaxReservedDescID+2, &config.ZoneConfig{RangeMaxBytes: 1 << 20})

	// Check our config.
	if sysCfg.NeedsSplit(neverSplits.Desc().StartKey, neverSplits.Desc().EndKey) {
		t.Fatal("System config says range needs to be split")
	}
	if !sysCfg.NeedsSplit(willSplit.Desc().StartKey, willSplit.Desc().EndKey) {
		t.Fatal("System config says range does not need to be split")
	}

	bq.MaybeAdd(neverSplits, roachpb.ZeroTimestamp)
	bq.MaybeAdd(willSplit, roachpb.ZeroTimestamp)

	if err := util.IsTrueWithin(func() bool {
		return atomic.LoadInt32(&testQueue.processed) == 3
	}, 250*time.Millisecond); err != nil {
		t.Error(err)
	}

	if pc := atomic.LoadInt32(&queued); pc != 3 {
		t.Errorf("expected queued count of 3; got %d", pc)
	}
}

// TestSplitQueueShouldQueue verifies shouldQueue method correctly
// combines splits in zone configs with the size of the range.
func TestSplitQueueShouldQueue(t *testing.T) {
	defer leaktest.AfterTest(t)()
	tc := testContext{}
	tc.Start(t)
	defer tc.Stop()

	// Set zone configs.
	config.TestingSetZoneConfig(2000, &config.ZoneConfig{RangeMaxBytes: 32 << 20})
	config.TestingSetZoneConfig(2002, &config.ZoneConfig{RangeMaxBytes: 32 << 20})

	// Despite faking the zone configs, we still need to have a gossip entry.
	if err := tc.gossip.AddInfoProto(gossip.KeySystemConfig, &config.SystemConfig{}, 0); err != nil {
		t.Fatal(err)
	}

	testCases := []struct {
		start, end roachpb.RKey
		bytes      int64
		shouldQ    bool
		priority   float64
	}{
		// No intersection, no bytes.
		{roachpb.RKeyMin, roachpb.RKey("/"), 0, false, 0},
		// Intersection in zone, no bytes.
		{keys.MakeTablePrefix(2001), roachpb.RKeyMax, 0, true, 1},
		// Already split at largest ID.
		{keys.MakeTablePrefix(2002), roachpb.RKeyMax, 0, false, 0},
		// Multiple intersections, no bytes.
		{roachpb.RKeyMin, roachpb.RKeyMax, 0, true, 1},
		// No intersection, max bytes.
		{roachpb.RKeyMin, roachpb.RKey("/"), 64 << 20, false, 0},
		// No intersection, max bytes+1.
		{roachpb.RKeyMin, roachpb.RKey("/"), 64<<20 + 1, true, 1},
		// No intersection, max bytes * 2.
		{roachpb.RKeyMin, roachpb.RKey("/"), 64 << 21, true, 2},
		// Intersection, max bytes +1.
		{keys.MakeTablePrefix(2000), roachpb.RKeyMax, 32<<20 + 1, true, 2},
		// Split needed at table boundary, but no zone config.
		{keys.MakeTablePrefix(2001), roachpb.RKeyMax, 32<<20 + 1, true, 1},
	}

	splitQ := newSplitQueue(nil, tc.gossip)

	cfg := tc.gossip.GetSystemConfig()
	if cfg == nil {
		t.Fatal("nil config")
	}

	for i, test := range testCases {
		if err := tc.rng.stats.SetMVCCStats(tc.rng.store.Engine(), engine.MVCCStats{KeyBytes: test.bytes}); err != nil {
			t.Fatal(err)
		}
		copy := *tc.rng.Desc()
		copy.StartKey = test.start
		copy.EndKey = test.end
		if err := tc.rng.setDesc(&copy); err != nil {
			t.Fatal(err)
		}
		shouldQ, priority := splitQ.shouldQueue(roachpb.ZeroTimestamp, tc.rng, cfg)
		if shouldQ != test.shouldQ {
			t.Errorf("%d: should queue expected %t; got %t", i, test.shouldQ, shouldQ)
		}
		if math.Abs(priority-test.priority) > 0.00001 {
			t.Errorf("%d: priority expected %f; got %f", i, test.priority, priority)
		}
	}
}

// TestGCQueueLookupGCPolicy verifies gc policy lookups.
func TestGCQueueLookupGCPolicy(t *testing.T) {
	defer leaktest.AfterTest(t)
	tc := testContext{}
	tc.Start(t)
	defer tc.Stop()

	zoneDefault := config.DefaultZoneConfig
	zoneTable1000 := &config.ZoneConfig{
		ReplicaAttrs:  []proto.Attributes{},
		RangeMinBytes: 1 << 10,
		RangeMaxBytes: 1 << 18,
		GC: &config.GCPolicy{
			TTLSeconds: 60 * 60, // 1 hour only
		},
	}
	zoneTable1002 := &config.ZoneConfig{
		ReplicaAttrs:  []proto.Attributes{},
		RangeMinBytes: 1 << 10,
		RangeMaxBytes: 1 << 18,
		// Note that there is no GC set here, so we should select the
		// hierarchical parent's GC policy; in this case, zoneConfig1.
	}

	// Add configs to testing helper.
	config.TestingSetZoneConfig(1000, zoneTable1000)
	config.TestingSetZoneConfig(1002, zoneTable1002)

	testCases := []struct {
		start, end proto.Key
		zoneConfig *config.ZoneConfig
		errStr     string
	}{
		{proto.KeyMin, proto.KeyMax, nil, "spans multiple ranges"},
		{keys.MakeTablePrefix(1000), keys.MakeTablePrefix(10002), nil, "spans multiple ranges"},
		{proto.KeyMin, proto.Key("a"), zoneDefault, ""},
		{keys.MakeTablePrefix(1000), keys.MakeTablePrefix(1001), zoneTable1000, ""},
		{keys.MakeTablePrefix(1001), keys.MakeTablePrefix(1002), zoneDefault, ""},
		{keys.MakeTablePrefix(1002), keys.MakeTablePrefix(1010), zoneTable1002, ""},
		{keys.MakeTablePrefix(1002), proto.KeyMax, zoneTable1002, ""},
		{keys.MakeTablePrefix(9999), proto.KeyMax, zoneDefault, ""},
	}

	gcQ := newGCQueue()
	for testNum, testCase := range testCases {
		rng := createRange(tc.store, proto.RangeID(testNum+1), testCase.start, testCase.end)

		gcPolicy, err := gcQ.lookupGCPolicy(rng)
		if testCase.errStr == "" {
			if err != nil {
				t.Errorf("#%d: error: %v", testNum, err)
				continue
			}
		} else if !testutils.IsError(err, testCase.errStr) {
			t.Errorf("#%d: expected err=%s, got %v", testNum, testCase.errStr, err)
			continue
		}
		if testCase.zoneConfig == nil {
			continue
		}
		if !reflect.DeepEqual(gcPolicy, testCase.zoneConfig.GC) {
			t.Errorf("#%d: mismatching GCPolicy, got: %+v, expected: %+v",
				testNum, gcPolicy, testCase.zoneConfig.GC)
		}
	}
}