Merge #144482 #145080

144482: kvserver/closedts: add TestPolicyRefresherWithLatencies r=arulajmani a=wenyihu6

This commit adds a new test, TestPolicyRefresherWithLatencies, which verifies
that the policy refresher correctly updates latency information using mocked
latency map data.

Resolves: #143888
Release note: none

145080: sql: fix ALTER TABLE SET LOGGED/UNLOGGED r=rafiss a=rafiss

This was recently added in b614b04, but for the notice to display correctly we must not return a nil planNode.

No release note since the bug has not been released.

informs #144727
Release note: None

Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

Author: 3 people

pkg/kv/kvserver/closedts/policyrefresher/policy_refresher_test.go

@@ -40,19 +40,25 @@ type mockSpanConfig struct {
 
 type mockReplica struct {
 	// Note that all fields below are protected by mu.
-	mu     syncutil.Mutex
-	conf   mockSpanConfig
-	policy ctpb.RangeClosedTimestampPolicy
+	mu             syncutil.Mutex
+	conf           mockSpanConfig
+	policy         ctpb.RangeClosedTimestampPolicy
+	furthestNodeID roachpb.NodeID
 }
 
-func (m *mockReplica) RefreshPolicy(_ map[roachpb.NodeID]time.Duration) {
+func (m *mockReplica) RefreshPolicy(latencies map[roachpb.NodeID]time.Duration) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	if m.conf.isGlobalRead {
-		m.policy = ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO
-	} else {
+	if !m.conf.isGlobalRead {
 		m.policy = ctpb.LAG_BY_CLUSTER_SETTING
+		return
+	}
+	latency, ok := latencies[m.furthestNodeID]
+	if !ok {
+		m.policy = ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO
+		return
 	}
+	m.policy = closedts.FindBucketBasedOnNetworkRTT(latency)
 }
 
 func (m *mockReplica) GetPolicy() ctpb.RangeClosedTimestampPolicy {
@@ -61,6 +67,12 @@ func (m *mockReplica) GetPolicy() ctpb.RangeClosedTimestampPolicy {
 	return m.policy
 }
 
+func (m *mockReplica) SetFurthestNodeID(nodeID roachpb.NodeID) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.furthestNodeID = nodeID
+}
+
 func (m *mockReplica) BlockReplica() (unblock func()) {
 	m.mu.Lock()
 	var once sync.Once
@@ -243,3 +255,65 @@ func TestPolicyRefresherOnLatencyIntervalUpdate(t *testing.T) {
 		return errors.New("expected latency update")
 	})
 }
+
+// TestPolicyRefresherWithLatencies tests that the policy refresher correctly
+// updates policies based on latency information from different nodes.
+func TestPolicyRefresherWithLatencies(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx := context.Background()
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+	st := cluster.MakeTestingClusterSettings()
+	closedts.LeadForGlobalReadsAutoTuneEnabled.Override(ctx, &st.SV, true)
+
+	// Create a policy refresher with test latencies.
+	latencies := map[roachpb.NodeID]time.Duration{
+		1: 10 * time.Millisecond,
+		2: 50 * time.Millisecond,
+		3: 90 * time.Millisecond,
+	}
+
+	r := &mockReplica{}
+	// Configure replica to not use global reads span config initially.
+	r.SetSpanConfig(false)
+
+	getLeaseholders := func() []Replica { return []Replica{r} }
+	getLatencies := func() map[roachpb.NodeID]time.Duration { return latencies }
+
+	pr := NewPolicyRefresher(stopper, st, getLeaseholders, getLatencies, nil)
+	require.NotNil(t, pr)
+	require.Nil(t, pr.getCurrentLatencies())
+
+	// Initially, the policy should be LAG_BY_CLUSTER_SETTING.
+	require.Equal(t, r.GetPolicy(), ctpb.LAG_BY_CLUSTER_SETTING)
+
+	// Enable global reads and set furthest node to 1. We should get
+	// LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO policy before the cache update.
+	r.SetSpanConfig(true)
+	r.SetFurthestNodeID(1)
+	pr.refreshPolicies([]Replica{r})
+	require.Equal(t, r.GetPolicy(), ctpb.LEAD_FOR_GLOBAL_READS_WITH_NO_LATENCY_INFO)
+
+	// Update latency cache and verify we now get <20ms policy for node 1.
+	pr.updateLatencyCache()
+	require.NotNil(t, pr.getCurrentLatencies())
+	pr.refreshPolicies([]Replica{r})
+	require.Equal(t, r.GetPolicy(), ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_20MS)
+
+	// Set furthest node to 2 and verify we get <60ms policy.
+	r.SetFurthestNodeID(2)
+	pr.refreshPolicies([]Replica{r})
+	require.Equal(t, r.GetPolicy(), ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_60MS)
+
+	// Set furthest node to 3 and verify we get <100ms policy.
+	r.SetFurthestNodeID(3)
+	pr.refreshPolicies([]Replica{r})
+	require.Equal(t, r.GetPolicy(), ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_LESS_THAN_100MS)
+
+	// Update node 3's latency to 300ms and verify we get >=300ms policy.
+	latencies[3] = 300 * time.Millisecond
+	pr.updateLatencyCache()
+	pr.refreshPolicies([]Replica{r})
+	require.Equal(t, r.GetPolicy(), ctpb.LEAD_FOR_GLOBAL_READS_LATENCY_EQUAL_OR_GREATER_THAN_300MS)
+}

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -4563,3 +4563,22 @@ statement ok
 DROP TYPE e1;
 
 subtest end
+
+subtest set_logged
+
+statement ok
+CREATE TABLE t_set_logged (a INT PRIMARY KEY);
+
+skipif config weak-iso-level-configs
+query T noticetrace
+ALTER TABLE t_set_logged SET LOGGED;
+----
+NOTICE: SET LOGGED is not supported and has no effect
+
+skipif config weak-iso-level-configs
+query T noticetrace
+ALTER TABLE t_set_logged SET UNLOGGED;
+----
+NOTICE: SET UNLOGGED is not supported and has no effect
+
+subtest end

pkg/sql/unimplemented.go

@@ -91,5 +91,5 @@ func (p *planner) AlterTableSetLogged(
 			operation,
 		),
 	)
-	return nil, nil
+	return &zeroNode{}, nil
 }