Merge #155033 #156594 #156876 #156935 #156942 #156943

155033: asim: use kvserver.LoadBasedRebalancingObjective r=wenyihu6 a=wenyihu6

Previously, LBRebalancingObjective was part of the simulation settings, using
QPS as the load rebalance objective. However, the cluster setting
kvserver.LoadBasedRebalancingObjective defaults to CPU rebalancing instead. This
commit updates asim to rely on the cluster setting, adopting the default cluster
setting configuration.

Epic: CRDB-55052
Release note: none 
Resolves: #154512

156594: backup: test adding/dropping database in cluster backup and restore r=jeffswenson a=msbutler

Epic: None

Release note: none

156876: rttanalysis: deflake BenchmarkJobs r=rafiss a=rafiss

The benchmark started failing since the code doesn't handle a nil table descriptor. The fix is to create a dummy descriptor in the job payload details.

fixes #151302
Release note: None

156935: logical: enable crud writer by default r=jeffswenson a=jeffswenson

This change enables the crud writer by default for IMMEDIATE and VALIDATED mode LDR jobs.

Release note: LDR now supports partial indexes and is tolerant of mismatched column ids in the source and destination tables.

Epic: CRDB-51533

156942: server: make mma Allocator calls more visible r=wenyihu6 a=tbg

They were kind of hidden before, and it has been
tripping me up every now and then for a while.
The problem was that NewAllocatorState returns
a private struct directly, so a bunch of the
calls later in the method do not show up when
you look at the call hierarchy for the Allocator
interface. This is now fixed.

Epic: CRDB-55052.

156943: workflow: backport stale action should mark PRs stale after 14 days r=RaduBerinde a=rail

Previously, the backport stale action marked PRs as stale after 21 days of inactivity. This change updates the configuration to mark PRs as stale after 14 days of inactivity, ensuring that stale backport PRs are identified and addressed more promptly.

Epic: none
Release note: none

Co-authored-by: wenyihu6 <wenyi@cockroachlabs.com>
Co-authored-by: Michael Butler <butler@cockroachlabs.com>
Co-authored-by: Rafi Shamim <rafi@cockroachlabs.com>
Co-authored-by: Jeff Swenson <jeffswenson@betterthannull.com>
Co-authored-by: Tobias Grieger <tobias.b.grieger@gmail.com>
Co-authored-by: Rail Aliiev <rail@iqchoice.com>

Author: 7 people

.github/workflows/backport-stale.yml

@@ -2,6 +2,7 @@ name: Mark stale backport requests
 
 on:
   schedule:
+  # Run at 10am UTC daily, except weekends
   - cron: "0 10 * * 1-4"
   workflow_dispatch:
 
@@ -12,17 +13,17 @@ jobs:
       issues: write
       pull-requests: write
     steps:
-    - uses: actions/stale@v3
+    - uses: actions/stale@v10
       with:
         operations-per-run: 1000
         repo-token: ${{ secrets.GITHUB_TOKEN }}
-        stale-issue-message: 'Blah'
-        stale-pr-message: 'Reminder: it has been 3 weeks please merge or close your backport!'
+        stale-issue-message: 'Ignored'
+        stale-pr-message: 'Reminder: it has been 2 weeks please merge or close your backport!'
         stale-issue-label: 'no-backport-issue-activity'
         stale-pr-label: 'no-backport-pr-activity'
         close-issue-label: 'X-stale'
         close-pr-label: 'X-stale'
-        days-before-pr-stale: 21
+        days-before-pr-stale: 14
         # Disable this for issues, by setting a very high bar
         days-before-issue-stale: 99999
         days-before-close: 99999

pkg/backup/backup_test.go

@@ -11198,3 +11198,94 @@ func TestBackupRestoreFunctionDependenciesRevisionHistory(t *testing.T) {
 
 	checkFunctions("test6")
 }
+
+func TestBackupRestoreDatabaseRevisionHistory(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	dataDir, dirCleanupFunc := testutils.TempDir(t)
+	defer dirCleanupFunc()
+
+	_, sqlDB, cleanupFn := backupRestoreTestSetupEmpty(t, singleNode, dataDir, InitManualReplication, base.TestClusterArgs{})
+	defer cleanupFn()
+
+	// Helper function to check if a database exists.
+	checkDatabase := func(dbName string, shouldExist bool) {
+		var count int
+		sqlDB.QueryRow(t, `SELECT count(*) FROM system.namespace WHERE name = $1 AND "parentID" = 0`, dbName).Scan(&count)
+		if shouldExist {
+			if count != 1 {
+				t.Fatalf("expected database %s to exist, but it doesn't", dbName)
+			}
+		} else {
+			if count != 0 {
+				t.Fatalf("expected database %s to not exist, but it does", dbName)
+			}
+		}
+	}
+
+	sqlDB.Exec(t, `CREATE DATABASE test_db`)
+
+	// T1: Full cluster backup with revision history.
+	backupPath := "nodelocal://1/database_revision_backup"
+	sqlDB.Exec(t, `BACKUP INTO $1 WITH revision_history`, backupPath)
+
+	// T2: Capture timestamp after backup (database exists).
+	var t2 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t2)
+
+	sqlDB.Exec(t, `DROP DATABASE test_db`)
+
+	checkDatabase("test_db", false)
+
+	// T3: Capture timestamp after dropping database.
+	var t3 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t3)
+
+	sqlDB.Exec(t, `CREATE DATABASE ephemeral`)
+
+	// T4: Capture timestamp the has the ephemeral database.
+	var t4 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t4)
+
+	sqlDB.Exec(t, `DROP DATABASE ephemeral`)
+
+	// T5: Capture timestamp after dropping ephemeral database.
+	var t5 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t5)
+
+	sqlDB.Exec(t, `BACKUP INTO LATEST IN $1 WITH revision_history`, backupPath)
+
+	// Choose one of the 4 restores to run, to reduce test runtime:
+	i := rand.Intn(4)
+
+	switch i {
+	case 0:
+		// Restore AOST t2 -> expect database to exist.
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t2)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("test_db", true)
+
+		sqlDB.Exec(t, `DROP DATABASE test_db`)
+	case 1:
+		// Restore AOST t3 -> expect database to not exist.
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t3)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("test_db", false)
+	case 2:
+		// Restore AOST t4 -> ephemeral db exists.
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t4)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("ephemeral", true)
+
+		sqlDB.Exec(t, `DROP DATABASE ephemeral`)
+	case 3:
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t5)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("ephemeral", false)
+	}
+}

pkg/bench/rttanalysis/BUILD.bazel

@@ -65,6 +65,7 @@ go_test(
         "//pkg/security/username",
         "//pkg/server",
         "//pkg/server/serverpb",
+        "//pkg/sql/catalog/descpb",
         "//pkg/testutils/pgurlutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",

pkg/bench/rttanalysis/jobs_test.go

@@ -14,13 +14,27 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 )
 
 func BenchmarkJobs(b *testing.B) { reg.Run(b) }
 func init() {
+	// Create a minimal table descriptor for the import job.
+	tableDesc := &descpb.TableDescriptor{
+		ID:            100,
+		ParentID:      1,
+		Name:          "benchmark_table",
+		FormatVersion: descpb.InterleavedFormatVersion,
+		Version:       1,
+	}
+
 	payloadBytes, err := protoutil.Marshal(&jobspb.Payload{
-		Details:       jobspb.WrapPayloadDetails(jobspb.ImportDetails{}),
+		Details: jobspb.WrapPayloadDetails(jobspb.ImportDetails{
+			Table: jobspb.ImportDetails_Table{
+				Desc: tableDesc,
+			},
+		}),
 		UsernameProto: username.RootUserName().EncodeProto(),
 	})
 	if err != nil {

pkg/crosscluster/logical/BUILD.bazel

@@ -150,6 +150,7 @@ go_test(
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/storageccl",
         "//pkg/clusterversion",
+        "//pkg/crosscluster",
         "//pkg/crosscluster/replicationtestutils",
         "//pkg/crosscluster/streamclient",
         "//pkg/crosscluster/streamclient/randclient",

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationtestutils"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient"
 	_ "github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient/randclient"
@@ -895,9 +896,6 @@ func TestRandomTables(t *testing.T) {
 	tc, s, runnerA, runnerB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
 	defer tc.Stopper().Stop(ctx)
 
-	// TODO(#148303): Remove this once the crud writer supports tables with array primary keys.
-	runnerA.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.immediate_mode_writer = 'legacy-kv'")
-
 	sqlA := s.SQLConn(t, serverutils.DBName("a"))
 
 	var tableName, streamStartStmt string
@@ -2897,8 +2895,14 @@ func TestGetWriterType(t *testing.T) {
 
 	t.Run("validated-mode", func(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
+
 		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Validated, st)
 		require.NoError(t, err)
+		require.Equal(t, sqlclustersettings.LDRWriterTypeCRUD, wt)
+
+		crosscluster.LogicalReplicationUDFWriterEnabled.Override(ctx, &st.SV, true)
+		wt, err = getWriterType(ctx, jobspb.LogicalReplicationDetails_Validated, st)
+		require.NoError(t, err)
 		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 	})
 

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -741,11 +741,17 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 func getWriterType(
 	ctx context.Context, mode jobspb.LogicalReplicationDetails_ApplyMode, settings *cluster.Settings,
 ) (sqlclustersettings.LDRWriterType, error) {
+	// TODO(jeffswenson): delete the kv and legacy sql ldr writers
 	switch mode {
 	case jobspb.LogicalReplicationDetails_Immediate:
 		return sqlclustersettings.LDRWriterType(sqlclustersettings.LDRImmediateModeWriter.Get(&settings.SV)), nil
 	case jobspb.LogicalReplicationDetails_Validated:
-		return sqlclustersettings.LDRWriterTypeSQL, nil
+		if crosscluster.LogicalReplicationUDFWriterEnabled.Get(&settings.SV) {
+			// If the UDF writer is enabled, fall back to the legacy SQL writer for
+			// validated mode.
+			return sqlclustersettings.LDRWriterTypeSQL, nil
+		}
+		return sqlclustersettings.LDRWriterTypeCRUD, nil
 	default:
 		return "", errors.Newf("unknown logical replication writer type: %s", mode)
 	}

pkg/kv/kvserver/asim/config/settings.go

@@ -29,8 +29,6 @@ const (
 	defaultSplitStatRetention      = 10 * time.Minute
 	defaultSeed                    = 42
 	defaultLBRebalancingInterval   = time.Minute
-	defaultLBRebalanceQPSThreshold = 0.1
-	defaultLBRebalancingObjective  = 0 // QPS
 )
 
 const DefaultNodeCPURateCapacityNanos = 8 * 1e9        // 8 vcpus
@@ -101,8 +99,6 @@ type SimulationSettings struct {
 	// SplitStatRetention is the duration which recorded load will be retained
 	// and factored into load based splitting decisions.
 	SplitStatRetention time.Duration
-	// LBRebalancingObjective is the load objective to balance.
-	LBRebalancingObjective int64
 	// LBRebalancingInterval controls how often the store rebalancer will
 	// consider opportunities for rebalancing.
 	LBRebalancingInterval time.Duration
@@ -141,7 +137,6 @@ func DefaultSimulationSettings() *SimulationSettings {
 		StateExchangeDelay:      defaultStateExchangeDelay,
 		SplitQPSThreshold:       defaultSplitQPSThreshold,
 		SplitStatRetention:      defaultSplitStatRetention,
-		LBRebalancingObjective:  defaultLBRebalancingObjective,
 		LBRebalancingInterval:   defaultLBRebalancingInterval,
 		ReplicateQueueEnabled:   true,
 		LeaseQueueEnabled:       true,

pkg/kv/kvserver/asim/state/impl.go

@@ -1402,6 +1402,8 @@ func (s *state) SetClusterSetting(Key string, Value interface{}) {
 	switch Key {
 	case "LBRebalancingMode":
 		kvserverbase.LoadBasedRebalancingMode.Override(context.Background(), &s.settings.ST.SV, kvserverbase.LBRebalancingMode(Value.(int64)))
+	case "LBRebalancingObjective":
+		kvserver.LoadBasedRebalancingObjective.Override(context.Background(), &s.settings.ST.SV, kvserver.LBRebalancingObjective(Value.(int64)))
 	default:
 		panic("other cluster settings not supported")
 	}

pkg/kv/kvserver/asim/storerebalancer/store_rebalancer.go

@@ -126,11 +126,11 @@ type simRebalanceObjectiveProvider struct {
 
 // Objective returns the current rebalance objective.
 func (s simRebalanceObjectiveProvider) Objective() kvserver.LBRebalancingObjective {
-	return kvserver.LBRebalancingObjective(s.settings.LBRebalancingObjective)
+	return kvserver.LoadBasedRebalancingObjective.Get(&s.settings.ST.SV)
 }
 
 func (src *storeRebalancerControl) scorerOptions() *allocatorimpl.LoadScorerOptions {
-	dim := kvserver.LBRebalancingObjective(src.settings.LBRebalancingObjective).ToDimension()
+	dim := kvserver.LoadBasedRebalancingObjective.Get(&src.settings.ST.SV).ToDimension()
 	return &allocatorimpl.LoadScorerOptions{
 		BaseScorerOptions: allocatorimpl.BaseScorerOptions{
 			IOOverload:    src.allocator.IOOverloadOptions(),
@@ -191,7 +191,7 @@ func (src *storeRebalancerControl) phasePrologue(
 		ctx, src.scorerOptions(),
 		hottestRanges(
 			s, src.storeID,
-			kvserver.LBRebalancingObjective(src.settings.LBRebalancingObjective).ToDimension(),
+			kvserver.LoadBasedRebalancingObjective.Get(&src.settings.ST.SV).ToDimension(),
 		),
 		kvserverbase.LoadBasedRebalancingMode.Get(&src.settings.ST.SV),
 	)